fix Partitions

Author: suxiaogang223

fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergMetadataJniScanner.java

@@ -38,13 +38,13 @@
  * Abstract class for Iceberg metadata scanner.
  */
 public abstract class IcebergMetadataJniScanner extends JniScanner {
+    protected static final Logger LOG = LoggerFactory.getLogger(IcebergMetadataJniScanner.class);
     protected final String[] requiredFields;
     protected final Table table;
     protected final String timezone;
     protected Iterator<?> reader; // reader is initialized in the initReader() method
 
     private static final String HADOOP_OPTION_PREFIX = "hadoop.";
-    private static final Logger LOG = LoggerFactory.getLogger(IcebergMetadataJniScanner.class);
     private final PreExecutionAuthenticator preExecutionAuthenticator;
     private final ClassLoader classLoader;
     private ColumnType[] requiredTypes;

fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergPartitionsJniScanner.java

@@ -24,6 +24,8 @@
 import org.apache.iceberg.ManifestFiles;
 import org.apache.iceberg.PartitionData;
 import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.types.Type;
@@ -41,10 +43,14 @@
 
 class IcebergPartitionsJniScanner extends IcebergMetadataJniScanner {
     private static final String NAME = "partitions";
+    private static final List<String> SCAN_COLUMNS = List.of("content", "partition", "file_size_in_bytes",
+            "record_count");
 
     private List<PartitionField> partitionFields;
     private Long lastUpdateTime;
+    private Long lastUpdateSnapshotId;
     private Integer specId;
+    private Schema schema;
     private GenericRecord reusedRecord;
 
     // A serializable bean that contains a bare minimum to read a manifest
@@ -58,16 +64,18 @@ public IcebergPartitionsJniScanner(int batchSize, Map<String, String> params) {
     @Override
     protected void initReader() throws IOException {
         this.specId = manifestBean.partitionSpecId();
-        this.lastUpdateTime = table.snapshot(manifestBean.snapshotId()) != null
-                ? table.snapshot(manifestBean.snapshotId()).timestampMillis()
-                : null;
+        Snapshot snapshot = table.snapshot(manifestBean.snapshotId());
+        this.lastUpdateTime = snapshot != null ? snapshot.timestampMillis() : null;
+        this.lastUpdateSnapshotId = snapshot != null ? snapshot.snapshotId() : null;
         this.partitionFields = getAllPartitionFields(table);
-        // TODO: Initialize the reused record with partition fields
-        // this.reusedRecord = GenericRecord.create(getResultType());
+        this.schema = table.schema();
+        this.reusedRecord = GenericRecord.create(getResultType());
         if (manifestBean.content() == ManifestContent.DATA) {
-            reader = ManifestFiles.read(manifestBean, table.io(), table.specs()).iterator();
+            reader = ManifestFiles.read(manifestBean, table.io(), table.specs()).select(SCAN_COLUMNS)
+                    .caseSensitive(false).iterator();
         } else {
-            reader = ManifestFiles.readDeleteManifest(manifestBean, table.io(), table.specs()).iterator();
+            reader = ManifestFiles.readDeleteManifest(manifestBean, table.io(), table.specs()).select(SCAN_COLUMNS)
+                    .caseSensitive(false).iterator();
         }
     }
 
@@ -83,25 +91,39 @@ protected Object getColumnValue(String columnName, Object row) {
             case "record_count":
                 return content == FileContent.DATA ? file.recordCount() : 0;
             case "file_count":
-                return content == FileContent.DATA ? 1L : 0L;
+                return content == FileContent.DATA ? 1 : 0;
             case "total_data_file_size_in_bytes":
                 return content == FileContent.DATA ? file.fileSizeInBytes() : 0;
             case "position_delete_record_count":
                 return content == FileContent.POSITION_DELETES ? file.recordCount() : 0;
             case "position_delete_file_count":
-                return content == FileContent.POSITION_DELETES ? 1L : 0L;
+                return content == FileContent.POSITION_DELETES ? 1 : 0;
             case "equality_delete_record_count":
                 return content == FileContent.EQUALITY_DELETES ? file.recordCount() : 0;
             case "equality_delete_file_count":
-                return content == FileContent.EQUALITY_DELETES ? 1L : 0L;
+                return content == FileContent.EQUALITY_DELETES ? 1 : 0;
             case "last_updated_at":
                 return lastUpdateTime;
+            case "last_updated_snapshot_id":
+                return lastUpdateSnapshotId;
             default:
-                throw new IllegalArgumentException(
-                        "Unrecognized column name " + columnName + " in Iceberg " + NAME + " metadata table");
+                LOG.warn("Unrecognized column name " + columnName + " in Iceberg " + NAME + " metadata table");
+                return null;
         }
     }
 
+    private Types.StructType getResultType() {
+        List<Types.NestedField> fields = new ArrayList<>();
+        for (PartitionField partitionField : partitionFields) {
+            int id = partitionField.fieldId();
+            String name = partitionField.name();
+            Type type = partitionField.transform().getResultType(schema.findType(partitionField.sourceId()));
+            Types.NestedField nestedField = Types.NestedField.optional(id, name, type);
+            fields.add(nestedField);
+        }
+        return Types.StructType.of(fields);
+    }
+
     private Object getPartitionValues(PartitionData partitionData) {
         List<Types.NestedField> fileFields = partitionData.getPartitionType().fields();
         Map<Integer, Integer> fieldIdToPos = new HashMap<>();

fe/fe-core/src/main/java/org/apache/doris/tablefunction/IcebergTableValuedFunction.java

@@ -61,7 +61,7 @@ public class IcebergTableValuedFunction extends MetadataTableValuedFunction {
     private static final ImmutableSet<String> PROPERTIES_SET = ImmutableSet.of(TABLE, QUERY_TYPE);
     private static final ImmutableSet<TIcebergQueryType> CAN_SPLIT_TABLE_TYPES = ImmutableSet.of(
             TIcebergQueryType.FILES, TIcebergQueryType.DATA_FILES, TIcebergQueryType.DELETE_FILES,
-            TIcebergQueryType.POSITION_DELETES);
+            TIcebergQueryType.PARTITIONS, TIcebergQueryType.POSITION_DELETES);
 
     private final TIcebergQueryType queryType;
     private final Map<String, String> hadoopProps;
@@ -175,6 +175,7 @@ private List<String> getSplits() {
         }
         switch (queryType) {
             case FILES:
+            case PARTITIONS:
                 return table.currentSnapshot().allManifests(table.io()).stream()
                         .map(ManifestFileBean::fromManifest).map(SerializationUtil::serializeToBase64)
                         .collect(Collectors.toList());