[BugFix] Fix ORC file metric construction in AddFilesProcedure

Signed-off-by: chrisyguo <511955993@qq.com>

Author: Guosmilesmile

fe/fe-core/src/main/java/com/starrocks/connector/iceberg/procedure/AddFilesProcedure.java

@@ -482,11 +482,11 @@ private Metrics extractOrcMetrics(IcebergTableProcedureContext context, Table ta
                 ColumnStatistics[] columnStats = orcReader.getStatistics();
 
                 // Extract statistics for each column
-                for (int colId = 0; colId < columnStats.length; colId++) {
+                for (int colId = 1; colId < columnStats.length; colId++) {
                     ColumnStatistics stats = columnStats[colId];
 
                     // Map ORC column to Iceberg field
-                    String columnName = getColumnNameFromOrcSchema(orcSchema, colId);
+                    String columnName = getColumnNameFromOrcSchema(orcSchema, colId - 1);
                     if (columnName != null) {
                         Types.NestedField field = schema.findField(columnName);
                         if (field != null) {

fe/fe-core/src/test/java/com/starrocks/connector/iceberg/procedure/AddFilesProcedureTest.java

@@ -40,6 +40,10 @@
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.UUIDUtil;
+import org.apache.orc.ColumnStatistics;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.TypeDescription;
 import org.apache.parquet.column.Encoding;
 import org.apache.parquet.column.statistics.Statistics;
 import org.apache.parquet.hadoop.ParquetFileReader;
@@ -208,7 +212,7 @@ public void testExecuteWithSourceTable(@Mocked Table table, @Mocked IcebergHiveC
                 () -> procedure.execute(context, args));
 
         // The exception should be about Hive table access, not "not implemented"
-        assertTrue(exception.getMessage().contains("Failed to access source table") || 
+        assertTrue(exception.getMessage().contains("Failed to access source table") ||
                    exception.getMessage().contains("not found") ||
                    exception.getMessage().contains("not a Hive table"));
     }
@@ -598,6 +602,95 @@ public ParquetFileReader open(HadoopInputFile ignored) {
         assertTrue(metrics.upperBounds().isEmpty());
     }
 
+
+    @Test
+    public void testExtractOrcMetrics(@Mocked Table table,
+                                      @Mocked IcebergHiveCatalog catalog,
+                                      @Mocked Reader orcReader) throws Exception {
+        AddFilesProcedure procedure = AddFilesProcedure.getInstance();
+        IcebergTableProcedureContext context =
+                new IcebergTableProcedureContext(catalog, table, null, null, HDFS_ENVIRONMENT, null, null);
+
+        Schema schema = new Schema(
+                Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+                Types.NestedField.optional(2, "name", Types.StringType.get())
+        );
+
+        FileStatus fileStatus = new FileStatus(2048L, false, 1, 0L, 0L, new Path("/tmp/data.orc"));
+
+        TypeDescription orcSchema = TypeDescription.createStruct()
+                .addField("id", TypeDescription.createInt())
+                .addField("name", TypeDescription.createString());
+
+
+        long totalRows = 100L;
+
+        // index 0: root struct statistics (should not be mapped to any column)
+        ColumnStatistics rootStats = Mockito.mock(ColumnStatistics.class);
+        Mockito.when(rootStats.getNumberOfValues()).thenReturn(totalRows);
+        Mockito.when(rootStats.hasNull()).thenReturn(false);
+
+        // index 1: "id" column statistics, 90 non-null values, has nulls
+        ColumnStatistics idStats = Mockito.mock(ColumnStatistics.class);
+        Mockito.when(idStats.getNumberOfValues()).thenReturn(90L);
+        Mockito.when(idStats.hasNull()).thenReturn(true);
+
+        // index 2: "name" column statistics, 80 non-null values, has nulls
+        ColumnStatistics nameStats = Mockito.mock(ColumnStatistics.class);
+        Mockito.when(nameStats.getNumberOfValues()).thenReturn(80L);
+        Mockito.when(nameStats.hasNull()).thenReturn(true);
+
+        ColumnStatistics[] allStats = new ColumnStatistics[] {rootStats, idStats, nameStats};
+
+        new Expectations() {
+            {
+                table.schema();
+                result = schema;
+
+                orcReader.getNumberOfRows();
+                result = totalRows;
+
+                orcReader.getSchema();
+                result = orcSchema;
+
+                orcReader.getStatistics();
+                result = allStats;
+            }
+        };
+
+        new MockUp<OrcFile>() {
+            @Mock
+            public Reader createReader(Path path, OrcFile.ReaderOptions opts) {
+                return orcReader;
+            }
+        };
+
+        java.lang.reflect.Method method = AddFilesProcedure.class.getDeclaredMethod("extractOrcMetrics",
+                IcebergTableProcedureContext.class, Table.class, FileStatus.class);
+        method.setAccessible(true);
+        Metrics metrics = (Metrics) method.invoke(procedure, context, table, fileStatus);
+
+        assertEquals(totalRows, metrics.recordCount());
+
+
+        assertNotNull(metrics.valueCounts());
+        assertTrue(metrics.valueCounts().containsKey(1));
+        assertEquals(90L, metrics.valueCounts().get(1));
+
+        assertTrue(metrics.valueCounts().containsKey(2));
+        assertEquals(80L, metrics.valueCounts().get(2));
+
+        assertNotNull(metrics.nullValueCounts());
+        assertTrue(metrics.nullValueCounts().containsKey(1));
+        assertEquals(10L, metrics.nullValueCounts().get(1));
+
+        assertTrue(metrics.nullValueCounts().containsKey(2));
+        assertEquals(20L, metrics.nullValueCounts().get(2));
+
+        // Only 2 columns should have valueCounts, rootStats should not be mapped to any column
+        assertEquals(2, metrics.valueCounts().size());
+    }
+
     private IcebergTableProcedureContext createMockContext(@Mocked Table table, @Mocked IcebergHiveCatalog catalog) {
         ConnectContext ctx = Mockito.mock(ConnectContext.class);
         AlterTableStmt stmt = Mockito.mock(AlterTableStmt.class);