nsivabalan commented on code in PR #19045:
URL: https://github.com/apache/hudi/pull/19045#discussion_r3611578075
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1212,9 +1222,110 @@ private void initializeFileGroups(HoodieTableMetaClient
dataMetaClient, Metadata
writer.appendBlock(block);
}
} catch (InterruptedException e) {
- throw new HoodieException(String.format("Failed to created fileGroup
%s for partition %s", fileGroupFileId, relativePartitionPath), e);
+ throw new HoodieException(String.format("Failed to created fileGroup
%s for partition %s", fileGroupFileId, relativePath), e);
}
- }, fileGroupFileIds.size());
+ }, fileGroupIdAndPath.size());
+
+ // For non-flat layouts, write .hoodie_partition_metadata at the logical
partition root now,
+ // before the HoodieAppendHandle path (which would otherwise create one
inside the bucket
+ // sub-dir). The AppendHandle skips marker creation at layout sub-paths via
+ // HoodieAppendHandle#isMDTLayoutSubPath, so this is the sole marker write
under bucketing.
+ // For the flat default we leave marker creation to the AppendHandle and
write nothing here —
+ // existing tables keep bit-identical behavior.
+ if (!FlatMDTLayout.LAYOUT_ID.equals(layout.getLayoutId())) {
+ for (String markerPath :
layout.getPartitionMarkerPaths(relativePartitionPath, fileGroupCount)) {
+ HoodiePartitionMetadata marker = new HoodiePartitionMetadata(
+ dataMetaClient.getStorage(),
+ instantTime,
+ new StoragePath(metadataWriteConfig.getBasePath()),
+ FSUtils.constructAbsolutePath(metadataWriteConfig.getBasePath(),
markerPath),
+ Option.empty());
+ marker.trySave();
+ }
+ }
+
+ // Persist layout state for this partition (skipped entirely for the flat
default so existing
+ // tables get the identical on-disk and properties layout as before). When
a non-flat layout is
+ // in use, readers consult this property to enumerate physical sub-paths
without an FS listing.
+ if (metadataMetaClient != null &&
!FlatMDTLayout.LAYOUT_ID.equals(layout.getLayoutId())) {
+ maybePersistLayoutOnMDTInit(layout);
+
metadataMetaClient.getTableConfig().addMetadataLayoutPartitionFileGroupCounts(
+ metadataMetaClient, Collections.singletonMap(relativePartitionPath,
fileGroupCount));
Review Comment:
if we persist the mdt layout, do we need to refresh the metadataMetaClient
or not required?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java:
##########
@@ -245,11 +245,19 @@ protected void init(HoodieRecord record) {
deltaWriteStat.setFileId(fileId);
Option<FileSlice> fileSliceOpt =
populateWriteStatAndFetchFileSlice(record, deltaWriteStat);
try {
- HoodiePartitionMetadata partitionMetadata = new
HoodiePartitionMetadata(storage, instantTime,
- new StoragePath(config.getBasePath()),
- FSUtils.constructAbsolutePath(config.getBasePath(), partitionPath),
- hoodieTable.getPartitionMetafileFormat());
- partitionMetadata.trySave();
+ // For MDT writes under a non-flat layout (e.g., sub-directory
bucketing), the physical
+ // partitionPath here is a layout sub-path (e.g. "record_index/0004")
and the marker must
+ // live at the logical partition root instead. The marker at the logical
root is written
+ // once at MDT initialization in
HoodieBackedTableMetadataWriter.initializeFileGroups;
+ // skipping here keeps partition discovery returning logical names
rather than per-bucket
+ // sub-paths. For the flat default and the data table this guard is a
no-op.
+ if (!isMDTLayoutSubPath(partitionPath)) {
Review Comment:
can we add isMetadataTable as first condition
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -383,6 +383,34 @@ public static final String getDefaultPayloadClassName() {
.sinceVersion("1.1.0")
.withDocumentation("This property when set, will define how two versions
of the record will be merged together when records are partially formed");
+ public static final ConfigProperty<String> METADATA_LAYOUT_CLASS =
ConfigProperty
+ .key("hoodie.metadata.layout.class")
+ .noDefaultValue()
+ .sinceVersion("1.3.0")
+ .withDocumentation("Fully-qualified class name of the
HoodieMetadataTableLayout implementation that organizes "
Review Comment:
can we call out that this will be optionally added only for metadata table
when bucketing is enabled.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -1518,27 +1518,98 @@ private static List<FileSlice>
getPartitionFileSlices(HoodieTableMetaClient meta
HoodieTableFileSystemView fsView = null;
try {
fsView = fileSystemView.orElseGet(() ->
getFileSystemViewForMetadataTable(metaClient));
- Stream<FileSlice> fileSliceStream;
+ List<String> physicalPartitions = resolvePhysicalPartitions(metaClient,
partition);
+ if (physicalPartitions.isEmpty()) {
+ return Collections.emptyList();
+ }
+ // Hoist timeline lookups out of the loop — they depend only on
metaClient and stay invariant.
+ String mergedAsOfInstant = null;
if (mergeFileSlices) {
- if
(metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().isPresent())
{
- fileSliceStream = fsView.getLatestMergedFileSlicesBeforeOrOn(
- // including pending compaction instant as the last instant so
that the finished delta commits
- // that start earlier than the compaction can be queried.
- partition,
metaClient.getActiveTimeline().filterCompletedAndCompactionInstants().lastInstant().get().requestedTime());
- } else {
+ if
(!metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().isPresent())
{
return Collections.emptyList();
}
- } else {
- fileSliceStream = fsView.getLatestFileSlices(partition);
+ // Including pending compaction instant as the last instant so that
the finished delta
+ // commits that start earlier than the compaction can be queried.
+ mergedAsOfInstant =
metaClient.getActiveTimeline().filterCompletedAndCompactionInstants()
+ .lastInstant().get().requestedTime();
}
- return
fileSliceStream.sorted(Comparator.comparing(FileSlice::getFileId)).collect(Collectors.toList());
+ List<FileSlice> all = new ArrayList<>();
+ for (String physical : physicalPartitions) {
Review Comment:
can we name the variable better. may be `physicalPartition`
##########
hudi-common/src/main/java/org/apache/hudi/metadata/SubDirBucketedMDTLayout.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.metadata;
+
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.exception.HoodieMetadataException;
+import org.apache.hudi.storage.StoragePath;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Opt-in MDT layout that distributes file groups into bucket sub-directories
so a
+ * single MDT partition directory does not exceed per-directory file-count
limits
+ * common on HDFS-style filesystems.
+ *
+ * <p>On-disk shape for {@code record_index} with {@code fileGroupCount=2500}
and
+ * {@code bucketSize=1000}:
+ * <pre>
+ * .hoodie/metadata/record_index/
+ * ├── .hoodie_partition_metadata ← single marker at the
partition root
+ * ├── 000000/
+ * │ ├── .record-index-0000-0_<instant>.log
+ * │ ├── ...
+ * │ └── .record-index-0999-0_<instant>.log
+ * ├── 000001/
+ * │ └── ...
+ * └── 000002/
+ * └── ... (partial)
+ * </pre>
+ *
+ * <p>Notes:
+ * <ul>
+ * <li>The fileId scheme is unchanged. The bucket for a given file group is
+ * recoverable from the file-group index, which is itself encoded in the
+ * fileId. No additional state is required to map a fileId to its
bucket.</li>
+ * <li>A single {@code .hoodie_partition_metadata} marker lives at the
logical
+ * MDT partition root. None are written inside bucket sub-directories. This
+ * preserves MDT-as-Hudi-table semantics so {@code
FSUtils.getAllPartitionPaths}
+ * on the MDT returns logical partition names rather than physical bucket
+ * paths.</li>
+ * <li>Reads enumerate the physical sub-paths via
+ * {@link #getPhysicalPartitions(String, int)} sourced from the MDT's
persisted
+ * {@code hoodie.metadata.layout.partition.file.group.counts} property. No
+ * filesystem listing is needed on the read path.</li>
+ * </ul>
+ *
+ * <p><b>Scope:</b> this initial implementation supports the non-partitioned
MDT
+ * partitions (files, column_stats, bloom_filters, expression_index,
+ * secondary_index, and the global RLI). Partitioned RLI is rejected up front:
if
+ * a writer enables this layout on a table whose RLI is in the partitioned
mode,
+ * {@link #getFileGroupRelativePath} throws {@link HoodieMetadataException}.
+ * Partitioned-RLI bucketing requires a different growth model and lands in a
+ * separate follow-up.
+ */
+public final class SubDirBucketedMDTLayout implements
HoodieMetadataTableLayout {
+
+ public static final String LAYOUT_ID = "subdir-bucketed";
+ public static final int DEFAULT_BUCKET_SIZE = 1000;
+
+ private final int bucketSize;
+
+ public SubDirBucketedMDTLayout() {
+ this(DEFAULT_BUCKET_SIZE);
+ }
+
+ public SubDirBucketedMDTLayout(int bucketSize) {
+ ValidationUtils.checkArgument(bucketSize > 0,
+ "SubDirBucketedMDTLayout bucketSize must be > 0, got " + bucketSize);
+ this.bucketSize = bucketSize;
Review Comment:
I thought we designed the new configs so that we can support diff bucket
sizes for diff MDT partitions.
If yes, can you help me understand how can just one bucket taste good
enough? Or we should remove from the documentation of the new config property,
and even not allow the bucket size to change once a merida table is
initialized.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -1518,27 +1518,98 @@ private static List<FileSlice>
getPartitionFileSlices(HoodieTableMetaClient meta
HoodieTableFileSystemView fsView = null;
try {
fsView = fileSystemView.orElseGet(() ->
getFileSystemViewForMetadataTable(metaClient));
- Stream<FileSlice> fileSliceStream;
+ List<String> physicalPartitions = resolvePhysicalPartitions(metaClient,
partition);
+ if (physicalPartitions.isEmpty()) {
+ return Collections.emptyList();
Review Comment:
can you help me understand when would we end up here?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMDTLayoutBucketing.scala:
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.functional
+
+import org.apache.hudi.DataSourceWriteOptions._
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.HoodieTimeline
+import org.apache.hudi.config.HoodieCleanConfig
+import org.apache.hudi.metadata.{FlatMDTLayout, HoodieTableMetadata,
MetadataPartitionType, SubDirBucketedMDTLayout}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.spark.sql.SaveMode
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertNotNull, assertTrue}
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
+
+import scala.collection.JavaConverters._
+
+/**
+ * Validates that the MDT layout SPI works end-to-end for the two OSS-shipped
implementations:
+ *
+ * - {@link FlatMDTLayout} — today's behavior, file groups directly under
each MDT partition.
+ * - {@link SubDirBucketedMDTLayout} — file groups grouped into bucket
sub-directories.
+ *
+ * The same workload is run under both layouts and the MDT contract is checked:
+ *
+ * - RLI lookups return identical results under both layouts.
+ * - Logical MDT partitions are discoverable as Hudi partitions regardless
of bucketing
+ * ({@code FSUtils.getAllPartitionPaths} returns {@code [files,
record_index, ...]}, NOT bucket
+ * paths). This is the central correctness property we are protecting.
+ * - Direct Spark queries on the MDT path return non-empty results under
both layouts.
+ * - When bucketing is enabled, the on-disk structure actually uses bucket
sub-directories.
+ */
+class TestMDTLayoutBucketing extends RecordLevelIndexTestBase {
+
+ /**
+ * @param layoutClass FQCN of the layout to test; null means do not override
(flat default).
+ * @param bucketSize Bucket size to set when using sub-directory bucketing.
Ignored otherwise.
+ */
+ private def layoutOpts(layoutClass: String, bucketSize: Int): Map[String,
String] = {
+ if (layoutClass == null) {
+ Map.empty
+ } else {
+ Map(
+ HoodieMetadataConfig.METADATA_LAYOUT_CLASS.key -> layoutClass,
+ HoodieMetadataConfig.METADATA_LAYOUT_BUCKET_SIZE.key ->
bucketSize.toString)
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array(
+ "org.apache.hudi.metadata.FlatMDTLayout",
+ "org.apache.hudi.metadata.SubDirBucketedMDTLayout"))
+ def testRecordLevelIndexWritesAndLookupsAcrossLayouts(layoutClass: String):
Unit = {
+ // Force a small bucket size so even a modest workload exercises >1
buckets under the bucketed
+ // layout. The flat layout ignores bucketSize.
+ val opts = commonOpts ++ layoutOpts(layoutClass, bucketSize = 2)
+
+ // Bootstrap MDT + RLI with an INSERT.
+ doWriteAndValidateDataAndRecordIndex(opts, INSERT_OPERATION_OPT_VAL,
SaveMode.Overwrite)
+ // A couple of UPSERTs to exercise reads against initialized file groups.
+ doWriteAndValidateDataAndRecordIndex(opts, UPSERT_OPERATION_OPT_VAL,
SaveMode.Append)
+ doWriteAndValidateDataAndRecordIndex(opts, UPSERT_OPERATION_OPT_VAL,
SaveMode.Append)
+
+ metaClient =
HoodieTableMetaClient.builder().setBasePath(basePath).setConf(storageConf).build()
+
+ // Open MDT metaClient to inspect persisted layout state.
+ val mdtBasePath = HoodieTableMetadata.getMetadataTableBasePath(basePath)
+ val mdtMetaClient = HoodieTableMetaClient.builder()
+ .setBasePath(mdtBasePath).setConf(storageConf).build()
+
+ if (layoutClass == classOf[FlatMDTLayout].getName) {
+ // Flat layout must not persist a layout class — the default is
implicit, and existing tables
+ // (with no layout property) must continue to behave identically.
+
assertFalse(mdtMetaClient.getTableConfig.getMetadataLayoutClass.isPresent,
+ "flat layout must not persist hoodie.metadata.layout.class")
+ } else {
+ assertTrue(mdtMetaClient.getTableConfig.getMetadataLayoutClass.isPresent,
+ "non-flat layout must persist hoodie.metadata.layout.class")
+ assertEquals(layoutClass,
mdtMetaClient.getTableConfig.getMetadataLayoutClass.get)
+
assertTrue(mdtMetaClient.getTableConfig.getMetadataLayoutPartitionFileGroupCounts.asScala.nonEmpty,
+ "non-flat layout must persist per-partition file-group counts")
+ }
+
+ // Central correctness property: partition discovery on the MDT must
return logical names
+ // regardless of bucketing.
+ val mdtPartitions = FSUtils.getAllPartitionPaths(
+ context, mdtMetaClient, /* assumeDatePartitioning */ false).asScala.toSet
+
assertTrue(mdtPartitions.contains(MetadataPartitionType.FILES.getPartitionPath),
+ s"MDT must expose files partition; got: $mdtPartitions")
+
assertTrue(mdtPartitions.contains(MetadataPartitionType.RECORD_INDEX.getPartitionPath),
+ s"MDT must expose record_index partition; got: $mdtPartitions")
+ // None of the returned partitions should look like a bucket sub-path (6
digits at the end).
+ val bucketLike = mdtPartitions.filter(p => p.matches(".*/[0-9]{6}$"))
+ assertTrue(bucketLike.isEmpty,
+ s"MDT partition discovery must not expose bucket sub-paths as logical
partitions; got bucket-like: $bucketLike")
+
+ // For the bucketed layout, verify the on-disk structure actually uses
sub-directories.
+ if (layoutClass == classOf[SubDirBucketedMDTLayout].getName) {
+ val recordIndexDir = new StoragePath(mdtBasePath,
MetadataPartitionType.RECORD_INDEX.getPartitionPath)
+ val children =
mdtMetaClient.getStorage.listDirectEntries(recordIndexDir).asScala
+ val bucketDirs = children.filter(_.isDirectory)
+ assertTrue(bucketDirs.nonEmpty,
+ s"bucketed layout must produce at least one bucket sub-directory under
record_index; got children=${children.map(_.getPath.getName)}")
+ // Each bucket dir must be %06d-formatted.
+ bucketDirs.foreach { d =>
+ val name = d.getPath.getName
+ assertTrue(name.matches("[0-9]{6}"),
+ s"bucket sub-directory name must be %06d-formatted, got: $name")
+ }
+ // Marker must NOT live inside a bucket dir — it must live at the
partition root.
+ bucketDirs.foreach { d =>
+ val markerInsideBucket = new StoragePath(d.getPath,
".hoodie_partition_metadata")
+ assertFalse(mdtMetaClient.getStorage.exists(markerInsideBucket),
+ s".hoodie_partition_metadata must not exist inside bucket dir
${d.getPath}")
+ }
+ val markerAtRoot = new StoragePath(recordIndexDir,
".hoodie_partition_metadata")
+ assertTrue(mdtMetaClient.getStorage.exists(markerAtRoot),
+ s".hoodie_partition_metadata must exist at the logical partition root:
$markerAtRoot")
+ }
+
+ // Direct Spark query against the MDT path must return at least one row
under either layout.
+ val mdtDf = spark.read.format("hudi").load(mdtBasePath)
+ val mdtCount = mdtDf.count()
+ assertTrue(mdtCount > 0L,
+ s"direct Spark scan on MDT path must return at least one row under
layout $layoutClass; got $mdtCount")
+ assertNotNull(mdtDf.schema.fieldNames, "MDT schema must resolve via Spark
datasource")
+ }
+
+ /**
+ * Long-running workload validating that MDT table services (compaction +
cleaning) execute
+ * cleanly when the MDT is using the sub-directory bucketing layout. With
the bucketed layout,
+ * `BaseHoodieCompactionPlanGenerator` and the cleaner's full-listing path
go through the file
+ * system view under the logical MDT partition name — earlier reviews
flagged that this could
+ * skip bucketed file groups entirely. This test forces both services to
fire repeatedly so any
+ * regression in that area surfaces as either zero compaction/clean instants
or an exception.
+ *
+ * Workload: 25 upsert commits with MDT compaction set to fire every 5 delta
commits and a tight
+ * cleaner-commits-retained so cleaning has work to do early.
+ */
+ @Test
+ def testMDTTableServicesWithBucketing(): Unit = {
+ val opts = commonOpts ++
+ layoutOpts(classOf[SubDirBucketedMDTLayout].getName, bucketSize = 2) ++
+ Map(
+ // MDT compaction every 5 delta commits, so a 25-commit run triggers
it multiple times.
+ HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key -> "5",
+ // Tight cleaner so cleaning has work to do on the data table; the MDT
cleaner is driven by
+ // the data table's cleaner policy.
+ HoodieCleanConfig.AUTO_CLEAN.key -> "true",
+ HoodieCleanConfig.ASYNC_CLEAN.key -> "false",
+ HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key -> "3")
+
+ // Bootstrap.
+ doWriteAndValidateDataAndRecordIndex(opts, INSERT_OPERATION_OPT_VAL,
SaveMode.Overwrite)
+ // 24 more commits → 25 commits total. Validate RLI after each so a stale
slice surfaces fast.
+ (1 to 24).foreach { _ =>
+ doWriteAndValidateDataAndRecordIndex(opts, UPSERT_OPERATION_OPT_VAL,
SaveMode.Append)
+ }
+
+ metaClient =
HoodieTableMetaClient.builder().setBasePath(basePath).setConf(storageConf).build()
+ val mdtBasePath = HoodieTableMetadata.getMetadataTableBasePath(basePath)
+ val mdtMetaClient = HoodieTableMetaClient.builder()
+ .setBasePath(mdtBasePath).setConf(storageConf).build()
+
+ // Layout must still report itself as bucketed after the long run.
+ assertEquals(classOf[SubDirBucketedMDTLayout].getName,
+ mdtMetaClient.getTableConfig.getMetadataLayoutClass.get,
+ "MDT must still be on the bucketed layout after a long workload")
+
+ // -------- MDT compaction must have fired at least once. --------
+ val mdtTimeline = mdtMetaClient.reloadActiveTimeline()
+ val mdtCompactionInstants =
mdtTimeline.filterCompletedInstants().getInstants.asScala
+ .count(_.getAction == HoodieTimeline.COMMIT_ACTION)
+ // With max.delta.commits=5 and ~25 delta commits on the MDT, expect at
least one compaction.
+ assertTrue(mdtCompactionInstants >= 1,
+ s"expected >= 1 completed MDT compaction (COMMIT_ACTION) after the run;
got $mdtCompactionInstants " +
+ s"on timeline ${mdtTimeline.getInstants.asScala.toList}")
+
+ // -------- Data table cleaning must have fired at least once. --------
+ val dataTimeline = metaClient.reloadActiveTimeline()
+ val dataCleanInstants =
dataTimeline.getCleanerTimeline.filterCompletedInstants().countInstants()
+ assertTrue(dataCleanInstants >= 1,
+ s"expected >= 1 completed data table cleaner instant after the run; got
$dataCleanInstants")
+
+ // -------- Bucket sub-dirs still present (compaction did not delete
them). --------
+ val recordIndexDir = new StoragePath(mdtBasePath,
MetadataPartitionType.RECORD_INDEX.getPartitionPath)
+ val bucketDirs =
mdtMetaClient.getStorage.listDirectEntries(recordIndexDir).asScala
+ .filter(_.isDirectory)
+ assertTrue(bucketDirs.nonEmpty,
+ "bucket sub-directories under record_index must still exist after
compaction + cleaning")
+ bucketDirs.foreach { d =>
+ val name = d.getPath.getName
+ assertTrue(name.matches("[0-9]{6}"),
+ s"bucket sub-directory name must be %06d-formatted, got: $name")
+ // After compaction, each bucket should contain at least one base file
(HFile) — otherwise
+ // compaction silently skipped this bucket, which is the regression
cshuo / hudi-agent flagged.
+ val bucketEntries =
mdtMetaClient.getStorage.listDirectEntries(d.getPath).asScala
+ val hfiles = bucketEntries.filter(e =>
e.getPath.getName.endsWith(".hfile"))
+ assertTrue(hfiles.nonEmpty,
+ s"bucket ${d.getPath.getName} contains no HFile after compaction;
entries=${bucketEntries.map(_.getPath.getName)}")
+ }
+
+ // -------- The marker invariant must still hold post-compaction. --------
+ bucketDirs.foreach { d =>
+ val markerInsideBucket = new StoragePath(d.getPath,
".hoodie_partition_metadata")
+ assertFalse(mdtMetaClient.getStorage.exists(markerInsideBucket),
+ s"compaction must not introduce a .hoodie_partition_metadata inside
${d.getPath}")
+ }
+ val markerAtRoot = new StoragePath(recordIndexDir,
".hoodie_partition_metadata")
+ assertTrue(mdtMetaClient.getStorage.exists(markerAtRoot),
+ "logical-root marker must still exist after compaction + cleaning")
+
+ // -------- Direct Spark scan on the MDT path still returns rows. --------
+ val mdtDf = spark.read.format("hudi").load(mdtBasePath)
+ assertTrue(mdtDf.count() > 0L,
+ "direct Spark scan on MDT must still return rows after compaction +
cleaning")
+ }
Review Comment:
Can we add one functional test? It's a clean end-to-end function test.
1. Start a data delay.
2. Parameters for COWMOR: insert in the first batch, update in the second
commit, and then update in the third commit.
3. Let's trigger a compaction for MOR delay in the fourth commit.
4. Let's trigger clustering after five commits.
5. Two more updates.
The cleaner also, let's set it to four commits. In the data delay active
timeline, we should see ingestion commits, compaction commits, replace commits,
things like that. Let's validate the contents of both ParalY files, partition
and callstats partition, and then we'll see.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
