[fix](multi-catalog)support the max compute partition prune

be/src/runtime/descriptors.cpp

@@ -195,7 +195,6 @@ MaxComputeTableDescriptor::MaxComputeTableDescriptor(const TTableDescriptor& tde
           _table(tdesc.mcTable.table),
           _access_key(tdesc.mcTable.access_key),
           _secret_key(tdesc.mcTable.secret_key),
-          _partition_spec(tdesc.mcTable.partition_spec),
           _public_access(tdesc.mcTable.public_access) {}
 
 MaxComputeTableDescriptor::~MaxComputeTableDescriptor() = default;

be/src/runtime/descriptors.h

@@ -240,7 +240,6 @@ class MaxComputeTableDescriptor : public TableDescriptor {
     const std::string table() const { return _table; }
     const std::string access_key() const { return _access_key; }
     const std::string secret_key() const { return _secret_key; }
-    const std::string partition_spec() const { return _partition_spec; }
     const std::string public_access() const { return _public_access; }
 
 private:
@@ -249,7 +248,6 @@ class MaxComputeTableDescriptor : public TableDescriptor {
     std::string _table;
     std::string _access_key;
     std::string _secret_key;
-    std::string _partition_spec;
     std::string _public_access;
 };
 

be/src/vec/exec/format/table/max_compute_jni_reader.cpp

@@ -38,10 +38,15 @@ class Block;
 namespace doris::vectorized {
 
 MaxComputeJniReader::MaxComputeJniReader(const MaxComputeTableDescriptor* mc_desc,
+                                         const TMaxComputeFileDesc& max_compute_params,
                                          const std::vector<SlotDescriptor*>& file_slot_descs,
                                          const TFileRangeDesc& range, RuntimeState* state,
                                          RuntimeProfile* profile)
-        : _file_slot_descs(file_slot_descs), _range(range), _state(state), _profile(profile) {
+        : _max_compute_params(max_compute_params),
+          _file_slot_descs(file_slot_descs),
+          _range(range),
+          _state(state),
+          _profile(profile) {
     _table_desc = mc_desc;
     std::ostringstream required_fields;
     std::ostringstream columns_types;
@@ -64,7 +69,7 @@ MaxComputeJniReader::MaxComputeJniReader(const MaxComputeTableDescriptor* mc_des
                                        {"access_key", _table_desc->access_key()},
                                        {"secret_key", _table_desc->secret_key()},
                                        {"project", _table_desc->project()},
-                                       {"partition_spec", _table_desc->partition_spec()},
+                                       {"partition_spec", _max_compute_params.partition_spec},
                                        {"table", _table_desc->table()},
                                        {"public_access", _table_desc->public_access()},
                                        {"start_offset", std::to_string(_range.start_offset)},

be/src/vec/exec/format/table/max_compute_jni_reader.h

@@ -54,6 +54,7 @@ class MaxComputeJniReader : public GenericReader {
 
 public:
     MaxComputeJniReader(const MaxComputeTableDescriptor* mc_desc,
+                        const TMaxComputeFileDesc& max_compute_params,
                         const std::vector<SlotDescriptor*>& file_slot_descs,
                         const TFileRangeDesc& range, RuntimeState* state, RuntimeProfile* profile);
 
@@ -69,6 +70,7 @@ class MaxComputeJniReader : public GenericReader {
 
 private:
     const MaxComputeTableDescriptor* _table_desc = nullptr;
+    const TMaxComputeFileDesc& _max_compute_params;
     const std::vector<SlotDescriptor*>& _file_slot_descs;
     const TFileRangeDesc& _range;
     RuntimeState* _state = nullptr;

be/src/vec/exec/scan/vfile_scanner.cpp

@@ -745,13 +745,13 @@ Status VFileScanner::_get_next_reader() {
         bool need_to_get_parsed_schema = false;
         switch (format_type) {
         case TFileFormatType::FORMAT_JNI: {
-            if (_real_tuple_desc->table_desc()->table_type() ==
-                ::doris::TTableType::type::MAX_COMPUTE_TABLE) {
-                const MaxComputeTableDescriptor* mc_desc =
-                        static_cast<const MaxComputeTableDescriptor*>(
-                                _real_tuple_desc->table_desc());
+            if (range.__isset.table_format_params &&
+                range.table_format_params.table_format_type == "max_compute") {
+                const auto* mc_desc = static_cast<const MaxComputeTableDescriptor*>(
+                        _real_tuple_desc->table_desc());
                 std::unique_ptr<MaxComputeJniReader> mc_reader = MaxComputeJniReader::create_unique(
-                        mc_desc, _file_slot_descs, range, _state, _profile);
+                        mc_desc, range.table_format_params.max_compute_params, _file_slot_descs,
+                        range, _state, _profile);
                 init_status = mc_reader->init_reader(_colname_to_value_range);
                 _cur_reader = std::move(mc_reader);
             } else if (range.__isset.table_format_params &&

fe/be-java-extensions/max-compute-scanner/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java

@@ -61,11 +61,11 @@ public class MaxComputeJniScanner extends JniScanner {
     private static final String START_OFFSET = "start_offset";
     private static final String SPLIT_SIZE = "split_size";
     private static final String PUBLIC_ACCESS = "public_access";
-    private final RootAllocator arrowAllocator = new RootAllocator(Integer.MAX_VALUE);
     private final Map<String, MaxComputeTableScan> tableScans = new ConcurrentHashMap<>();
     private final String region;
     private final String project;
     private final String table;
+    private RootAllocator arrowAllocator;
     private PartitionSpec partitionSpec;
     private Set<String> partitionColumns;
     private MaxComputeTableScan curTableScan;
@@ -171,9 +171,14 @@ public void open() throws IOException {
             partitionColumns = session.getSchema().getPartitionColumns().stream()
                     .map(Column::getName)
                     .collect(Collectors.toSet());
-            List<Column> maxComputeColumns = new ArrayList<>(readColumns);
-            maxComputeColumns.removeIf(e -> partitionColumns.contains(e.getName()));
-            curReader = session.openArrowRecordReader(start, totalRows, maxComputeColumns, arrowAllocator);
+            List<Column> pushDownColumns = new ArrayList<>(readColumns);
+            pushDownColumns.removeIf(e -> partitionColumns.contains(e.getName()));
+            if (pushDownColumns.isEmpty() && !partitionColumns.isEmpty()) {
+                // query columns required non-null, when query partition table
+                pushDownColumns.add(session.getSchema().getColumn(0));
+            }
+            arrowAllocator = new RootAllocator(Integer.MAX_VALUE);
+            curReader = session.openArrowRecordReader(start, totalRows, pushDownColumns, arrowAllocator);
             remainBatchRows = totalRows;
         } catch (TunnelException e) {
             if (retryCount > 0 && e.getErrorMsg().contains("TableModified")) {
@@ -256,7 +261,8 @@ public void close() throws IOException {
         startOffset = -1;
         splitSize = -1;
         if (curReader != null) {
-            arrowAllocator.releaseBytes(arrowAllocator.getAllocatedMemory());
+            arrowAllocator.close();
+            arrowAllocator = null;
             curReader.close();
             curReader = null;
         }
@@ -281,15 +287,25 @@ protected int getNext() throws IOException {
     private int readVectors(int expectedRows) throws IOException {
         VectorSchemaRoot batch;
         int curReadRows = 0;
-        while (curReadRows < expectedRows && (batch = curReader.read()) != null) {
+        while (curReadRows < expectedRows) {
+            batch = curReader.read();
+            if (batch == null) {
+                break;
+            }
             try {
                 List<FieldVector> fieldVectors = batch.getFieldVectors();
                 int batchRows = 0;
                 for (FieldVector column : fieldVectors) {
+                    Integer readColumnId = readColumnsToId.get(column.getName());
+                    if (readColumnId == null) {
+                        // use for partition if no column need to read.
+                        batchRows = column.getValueCount();
+                        continue;
+                    }
                     columnValue.reset(column);
                     batchRows = column.getValueCount();
                     for (int j = 0; j < batchRows; j++) {
-                        appendData(readColumnsToId.get(column.getName()), columnValue);
+                        appendData(readColumnId, columnValue);
                     }
                 }
                 if (partitionSpec != null) {
@@ -305,6 +321,8 @@ private int readVectors(int expectedRows) throws IOException {
                     }
                 }
                 curReadRows += batchRows;
+            } catch (Exception e) {
+                throw new RuntimeException("Fail to read arrow data, reason: " + e.getMessage(), e);
             } finally {
                 batch.close();
             }

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

@@ -26,6 +26,7 @@
 import org.apache.doris.catalog.TableIf.TableType;
 import org.apache.doris.catalog.Type;
 import org.apache.doris.catalog.external.HMSExternalTable;
+import org.apache.doris.catalog.external.MaxComputeExternalTable;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
@@ -37,6 +38,7 @@
 import org.apache.doris.common.util.OrderByPair;
 import org.apache.doris.datasource.CatalogIf;
 import org.apache.doris.datasource.HMSExternalCatalog;
+import org.apache.doris.datasource.MaxComputeExternalCatalog;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.ShowResultSetMetaData;
@@ -125,7 +127,8 @@ public void analyze(Analyzer analyzer) throws UserException {
         }
 
         DatabaseIf db = catalog.getDbOrAnalysisException(dbName);
-        TableIf table = db.getTableOrMetaException(tblName, Table.TableType.OLAP, TableType.HMS_EXTERNAL_TABLE);
+        TableIf table = db.getTableOrMetaException(tblName, Table.TableType.OLAP,
+                    TableType.HMS_EXTERNAL_TABLE, TableType.MAX_COMPUTE_EXTERNAL_TABLE);
 
         if (table instanceof HMSExternalTable) {
             if (((HMSExternalTable) table).isView()) {
@@ -137,6 +140,13 @@ public void analyze(Analyzer analyzer) throws UserException {
             return;
         }
 
+        if (table instanceof MaxComputeExternalTable) {
+            if (((MaxComputeExternalTable) table).getOdpsTable().getPartitions().isEmpty()) {
+                throw new AnalysisException("Table " + tblName + " is not a partitioned table");
+            }
+            return;
+        }
+
         table.readLock();
         try {
             // build proc path
@@ -169,7 +179,8 @@ public void analyzeImpl(Analyzer analyzer) throws UserException {
         }
 
         // disallow unsupported catalog
-        if (!(catalog.isInternalCatalog() || catalog instanceof HMSExternalCatalog)) {
+        if (!(catalog.isInternalCatalog() || catalog instanceof HMSExternalCatalog
+                || catalog instanceof MaxComputeExternalCatalog)) {
             throw new AnalysisException(String.format("Catalog of type '%s' is not allowed in ShowPartitionsStmt",
                 catalog.getType()));
         }