[BugFix] fix iceberg transform compact bug (backport #62697) (#62766)

Signed-off-by: SevenJ <[email protected]>

Author: Wenjun7J

be/src/runtime/descriptors.cpp

@@ -239,6 +239,9 @@ IcebergTableDescriptor::IcebergTableDescriptor(const TTableDescriptor& tdesc, Ob
     } else {
         _source_column_names = tdesc.icebergTable.partition_column_names; //to compat with lower fe, set this also
         _partition_column_names = tdesc.icebergTable.partition_column_names;
+        for ([[maybe_unused]] const auto& _ : tdesc.icebergTable.partition_column_names) {
+            _transform_exprs.emplace_back("identity"); //to compat with lower fe, set this also
+        }
     }
 }
 

be/src/runtime/iceberg_table_sink.cpp

@@ -102,6 +102,18 @@ Status IcebergTableSink::decompose_to_pipeline(pipeline::OpFactories prev_operat
     } else {
         auto source_column_index = iceberg_table_desc->partition_source_index_in_schema();
         partition_expr = iceberg_table_desc->get_partition_exprs();
+
+        //for 3.5 fe -> 4.0 be compact, try to set this.
+        if (partition_expr.empty()) {
+            auto output_expr = this->get_output_expr();
+            for (const auto& index : source_column_index) {
+                if (index < 0 || index >= this->get_output_expr().size()) {
+                    return Status::InternalError(fmt::format("Invalid partition index: {}", index));
+                }
+                partition_expr.push_back(output_expr[index]);
+            }
+        }
+
         int idx = 0;
         for (auto& part_expr : partition_expr) {
             int index = source_column_index[idx];