[feature](merge-cloud) Add show data warehouse stmt

fe/fe-core/src/main/cup/sql_parser.cup

@@ -4288,13 +4288,13 @@ show_param ::=
         RESULT = new ShowDataSkewStmt(table_ref);
     :}
     /* Show data statement: used to show data size of specified range */
-     | KW_DATA order_by_clause:orderByClause
+     | KW_DATA order_by_clause:orderByClause opt_properties:prop
     {:
-        RESULT = new ShowDataStmt(null, orderByClause);
+        RESULT = new ShowDataStmt(null, orderByClause, prop);
     :}
     | KW_DATA KW_FROM table_name:dbTblName order_by_clause:orderByClause
     {:
-        RESULT = new ShowDataStmt(dbTblName, orderByClause);
+        RESULT = new ShowDataStmt(dbTblName, orderByClause, null);
     :}
     | opt_tmp:tmp KW_PARTITIONS KW_FROM table_name:tblName opt_wild_where order_by_clause:orderByClause limit_clause: limitClause
     {:

fe/fe-core/src/main/java/org/apache/doris/analysis/ShowDataStmt.java

@@ -75,6 +75,13 @@ public class ShowDataStmt extends ShowStmt {
                     .addColumn(new Column("RemoteSize", ScalarType.createVarchar(30)))
                     .build();
 
+    private static final ShowResultSetMetaData SHOW_WAREHOUSE_DATA_META_DATA =
+            ShowResultSetMetaData.builder()
+                    .addColumn(new Column("DBName", ScalarType.createVarchar(20)))
+                    .addColumn(new Column("DataSize", ScalarType.createVarchar(20)))
+                    .addColumn(new Column("RecycleSize", ScalarType.createVarchar(20)))
+                    .build();
+
     private static final ShowResultSetMetaData SHOW_INDEX_DATA_META_DATA =
             ShowResultSetMetaData.builder()
                     .addColumn(new Column("TableName", ScalarType.createVarchar(20)))
@@ -101,15 +108,78 @@ public class ShowDataStmt extends ShowStmt {
     private List<OrderByElement> orderByElements;
     private List<OrderByPair> orderByPairs;
 
-    public ShowDataStmt(TableName tableName, List<OrderByElement> orderByElements) {
+    private final Map<String, String> properties;
+
+    private static final String WAREHOUSE = "entire_warehouse";
+    private static final String DB_LIST = "db_names";
+
+    public ShowDataStmt(TableName tableName, List<OrderByElement> orderByElements, Map<String, String> properties) {
         this.tableName = tableName;
         this.totalRows = Lists.newArrayList();
         this.orderByElements = orderByElements;
+        this.properties = properties;
     }
 
     @Override
     public void analyze(Analyzer analyzer) throws UserException {
         super.analyze(analyzer);
+        if (properties != null) {
+            String value = properties.get(WAREHOUSE);
+            if (value != null && value.equals("true")) {
+                List<String> dbList = null;
+                String dbNames = properties.get(DB_LIST);
+                if (dbNames != null) {
+                    dbList = Arrays.asList(dbNames.split(","));
+                }
+                Map<String, Long> dbToDataSize = Env.getCurrentInternalCatalog().getUsedDataQuota();
+                Map<Long, Pair<Long, Long>> dbToRecycleSize = Env.getCurrentRecycleBin().getDbToRecycleSize();
+                Long total = 0L;
+                Long totalRecycleSize = 0L;
+                if (dbList == null) {
+                    for (Map.Entry<String, Long> pair : dbToDataSize.entrySet()) {
+                        Database db = Env.getCurrentInternalCatalog().getDbNullable(pair.getKey());
+                        if (db == null) {
+                            continue;
+                        }
+                        Long recycleSize = dbToRecycleSize.getOrDefault(db.getId(), Pair.of(0L, 0L)).first;
+                        List<String> result = Arrays.asList(db.getName(),
+                                String.valueOf(pair.getValue()), String.valueOf(recycleSize));
+                        totalRows.add(result);
+                        total += pair.getValue();
+                        totalRecycleSize += recycleSize;
+                        dbToRecycleSize.remove(db.getId());
+                    }
+
+                    // Append left database in recycle bin
+                    for (Map.Entry<Long, Pair<Long, Long>> entry : dbToRecycleSize.entrySet()) {
+                        List<String> result = Arrays.asList("NULL:" + entry.getKey(),
+                                "0", String.valueOf(entry.getValue().first));
+                        totalRows.add(result);
+                        totalRecycleSize += entry.getValue().first;
+                    }
+                } else {
+                    for (String databaseName : Env.getCurrentInternalCatalog().getDbNames()) {
+                        Database db = Env.getCurrentInternalCatalog().getDbNullable(databaseName);
+                        if (db == null) {
+                            continue;
+                        }
+                        if (!dbList.contains(db.getName())) {
+                            continue;
+                        }
+                        Long recycleSize = dbToRecycleSize.getOrDefault(db.getId(), Pair.of(0L, 0L)).first;
+                        Long dataSize = dbToDataSize.getOrDefault(databaseName, 0L);
+                        List<String> result =
+                                Arrays.asList(db.getName(), String.valueOf(dataSize), String.valueOf(recycleSize));
+                        totalRows.add(result);
+                        total += dataSize;
+                        totalRecycleSize += recycleSize;
+                    }
+                }
+                List<String> result = Arrays.asList("total", String.valueOf(total), String.valueOf(totalRecycleSize));
+                totalRows.add(result);
+                return;
+            }
+        }
         dbName = analyzer.getDefaultDb();
         if (Strings.isNullOrEmpty(dbName)) {
             getAllDbStats();

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -3564,4 +3564,12 @@ public void replayAutoIncrementIdUpdateLog(AutoIncrementIdUpdateLog log) throws
     public boolean enableAutoAnalyze() {
         return true;
     }
+
+    public Map<String, Long> getUsedDataQuota() {
+        Map<String, Long> dbToDataSize = new TreeMap<>();
+        for (Database db : this.idToDb.values()) {
+            dbToDataSize.put(db.getFullName(), db.getUsedDataQuotaWithLock());
+        }
+        return dbToDataSize;
+    }
 }

fe/fe-core/src/test/java/org/apache/doris/analysis/ShowDataStmtTest.java

@@ -139,7 +139,7 @@ public void setUp() throws UserException {
 
     @Test
     public void testNormal() throws AnalysisException, UserException {
-        ShowDataStmt stmt = new ShowDataStmt(null, null);
+        ShowDataStmt stmt = new ShowDataStmt(null, null, null);
         stmt.analyze(analyzer);
         Assert.assertEquals("SHOW DATA", stmt.toString());
         Assert.assertEquals(4, stmt.getMetaData().getColumnCount());
@@ -151,15 +151,15 @@ public void testNormal() throws AnalysisException, UserException {
         OrderByElement orderByElementTwo = new OrderByElement(slotRefTwo, false, false);
 
         stmt = new ShowDataStmt(new TableName(InternalCatalog.INTERNAL_CATALOG_NAME, "testDb", "test_tbl"),
-                Arrays.asList(orderByElementOne, orderByElementTwo));
+                Arrays.asList(orderByElementOne, orderByElementTwo), null);
         stmt.analyze(analyzer);
         Assert.assertEquals(
                 "SHOW DATA FROM `testDb`.`test_tbl` ORDER BY `ReplicaCount` DESC, `Size` DESC",
                 stmt.toString());
         Assert.assertEquals(6, stmt.getMetaData().getColumnCount());
         Assert.assertEquals(true, stmt.hasTable());
 
-        stmt = new ShowDataStmt(null, Arrays.asList(orderByElementOne, orderByElementTwo));
+        stmt = new ShowDataStmt(null, Arrays.asList(orderByElementOne, orderByElementTwo), null);
         stmt.analyze(analyzer);
         Assert.assertEquals("SHOW DATA ORDER BY `ReplicaCount` DESC, `Size` DESC",
                 stmt.toString());

regression-test/suites/show_p0/test_show_data_warehouse.groovy

@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_show_data_warehouse") {
+  sql """ DROP DATABASE IF EXISTS SHOW_DATA_1; """
+  sql """ DROP DATABASE IF EXISTS SHOW_DATA_2; """
+  sql """ CREATE DATABASE SHOW_DATA_1; """
+  sql """ CREATE DATABASE SHOW_DATA_2; """
+
+  sql """ USE SHOW_DATA_1; """
+
+  sql """ CREATE TABLE `table` (
+    `siteid` int(11) NOT NULL COMMENT "",
+    `citycode` int(11) NOT NULL COMMENT "",
+    `userid` int(11) NOT NULL COMMENT "",
+    `pv` int(11) NOT NULL COMMENT ""
+  ) ENGINE=OLAP
+  DUPLICATE KEY(`siteid`)
+  COMMENT "OLAP"
+  DISTRIBUTED BY HASH(`siteid`) BUCKETS 1; """
+
+  sql """ insert into `table` values
+        (9,10,11,12),
+        (9,10,11,12),
+        (1,2,3,4),
+        (13,21,22,16),
+        (13,14,15,16),
+        (17,18,19,20),
+        (1,2,3,4),
+        (13,21,22,16),
+        (13,14,15,16),
+        (17,18,19,20),
+        (5,6,7,8),
+        (5,6,7,8); """
+
+  sql """ USE SHOW_DATA_2; """
+
+  sql """ CREATE TABLE `table` (
+  `siteid` int(11) NOT NULL COMMENT "",
+  `citycode` int(11) NOT NULL COMMENT "",
+  `userid` int(11) NOT NULL COMMENT "",
+  `pv` int(11) NOT NULL COMMENT ""
+  ) ENGINE=OLAP
+  DUPLICATE KEY(`siteid`)
+  COMMENT "OLAP"
+  DISTRIBUTED BY HASH(`siteid`) BUCKETS 1; """
+
+  sql """ insert into `table` values
+        (9,10,11,12),
+        (9,10,11,12),
+        (1,2,3,4),
+        (13,21,22,16),
+        (13,14,15,16); """
+
+  // wait for heartbeat
+
+  long start = System.currentTimeMillis()
+  long dataSize = 0
+  long current = -1
+  do {
+    current = System.currentTimeMillis()
+    def res = sql """ show data properties("entire_warehouse"="true","db_names"="SHOW_DATA_1"); """
+    for (row : res) {
+      print row
+      if (row[0].toString() == "SHOW_DATA_1") {
+        dataSize = row[1].toInteger()
+      }
+    }
+    sleep(2000)
+  } while (dataSize == 0 && current - start < 600000)
+
+  qt_show_1 """ show data properties("entire_warehouse"="true","db_names"="SHOW_DATA_1"); """
+
+  qt_show_2 """ show data properties("entire_warehouse"="true","db_names"="SHOW_DATA_2"); """
+
+  qt_show_3 """ show data properties("entire_warehouse"="true","db_names"="SHOW_DATA_1,SHOW_DATA_2"); """
+
+  def result = sql """show data properties("entire_warehouse"="true")"""
+
+  assertTrue(result.size() >= 3)
+
+  sql """ DROP DATABASE IF EXISTS SHOW_DATA_1; """
+  result = sql """show data properties("entire_warehouse"="true")"""
+  assertTrue(result.size() > 0)
+  for (row : result) {
+    if (row[0].toString().equalsIgnoreCase("total")) {
+      assertTrue(row[2].toInteger() > 0)
+    }
+  }
+
+
+}