Fix instant file name path retrieval in Hudi Active Timeline

plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/model/HudiReplaceCommitMetadata.java

@@ -13,28 +13,34 @@
  */
 package io.trino.plugin.hudi.model;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 import static com.google.common.base.MoreObjects.toStringHelper;
+import static java.util.Objects.requireNonNull;
 
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class HudiReplaceCommitMetadata
 {
-    private Map<String, List<String>> partitionToReplaceFileIds;
-    private Boolean compacted;
+    private final Map<String, List<String>> partitionToReplaceFileIds;
+    private final boolean compacted;
 
-    // for ser/deser
-    public HudiReplaceCommitMetadata()
+    @JsonCreator
+    public HudiReplaceCommitMetadata(
+            @JsonProperty("partitionToReplaceFileIds") Map<String, List<String>> partitionToReplaceFileIds,
+            @JsonProperty("compacted") boolean compacted)
     {
-        partitionToReplaceFileIds = ImmutableMap.of();
-        compacted = false;
+        this.partitionToReplaceFileIds = ImmutableMap.copyOf(requireNonNull(partitionToReplaceFileIds, "partitionToReplaceFileIds is null"));
+        this.compacted = compacted;
     }
 
     public Map<String, List<String>> getPartitionToReplaceFileIds()
@@ -54,13 +60,14 @@ public boolean equals(Object o)
 
         HudiReplaceCommitMetadata that = (HudiReplaceCommitMetadata) o;
 
-        return compacted.equals(that.compacted);
+        return partitionToReplaceFileIds.equals(that.partitionToReplaceFileIds) &&
+               compacted == that.compacted;
     }
 
     @Override
     public int hashCode()
     {
-        return compacted.hashCode();
+        return Objects.hash(partitionToReplaceFileIds, compacted);
     }
 
     public static <T> T fromBytes(byte[] bytes, ObjectMapper objectMapper, Class<T> clazz)

plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/timeline/HudiActiveTimeline.java

@@ -85,7 +85,11 @@ public Optional<byte[]> readCompactionPlanAsBytes(HudiInstant instant)
 
     private Location getInstantFileNamePath(String fileName)
     {
-        return Location.of(fileName.contains(SCHEMA_COMMIT_ACTION) ? metaClient.getSchemaFolderName() : metaClient.getMetaPath().path()).appendPath(fileName);
+        Location metaPath = metaClient.getMetaPath();
+        if (fileName.contains(SCHEMA_COMMIT_ACTION)) {
+            return metaPath.appendPath(HudiTableMetaClient.SCHEMA_FOLDER_NAME).appendPath(fileName);
+        }
+        return metaPath.appendPath(fileName);
     }
 
     private Optional<byte[]> readDataFromPath(Location detailPath)

testing/trino-product-tests/src/main/java/io/trino/tests/product/hudi/TestHudiSparkCompatibility.java

@@ -320,6 +320,32 @@ public void testTimelineTableRedirect()
         }
     }
 
+    @Test(groups = {HUDI, PROFILE_SPECIFIC_TESTS})
+    public void testReadCopyOnWriteTableWithReplaceCommits()
+    {
+        String tableName = "test_hudi_cow_replace_commits_select_" + randomNameSuffix();
+
+        onHudi().executeQuery("CREATE TABLE default." + tableName +
+                              "(id bigint, name string, ts bigint)" +
+                              "USING hudi " +
+                              "TBLPROPERTIES (" +
+                              " type = 'cow'," +
+                              " primaryKey = 'id'," +
+                              " preCombineField = 'ts'," +
+                              " hoodie.clustering.inline = 'true'," +
+                              " hoodie.clustering.inline.max.commits = '1')" +
+                              "LOCATION 's3://" + bucketName + "/" + tableName + "'");
+
+        try {
+            onHudi().executeQuery("INSERT INTO default." + tableName + " VALUES (1, 'a1', 1000), (2, 'a2', 2000)");
+            assertThat(onTrino().executeQuery("SELECT id, name FROM hudi.default." + tableName))
+                    .containsOnly(row(1, "a1"), row(2, "a2"));
+        }
+        finally {
+            onHudi().executeQuery("DROP TABLE default." + tableName);
+        }
+    }
+
     private void createNonPartitionedTable(String tableName, String tableType)
     {
         onHudi().executeQuery(format(