Skip to content

Commit 4bafe5a

Browse files
wombatu-kunwombatu-kun
authored
test(trino): de-flake TestHudi*FileOperations by asserting only synchronous reads (apache#19004)
1 parent 23fe7bf commit 4bafe5a

3 files changed

Lines changed: 70 additions & 61 deletions

File tree

‎hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java‎

Lines changed: 50 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
import com.google.common.collect.ImmutableMap;
1818
import com.google.common.collect.ImmutableMultiset;
1919
import com.google.common.collect.Multiset;
20+
import io.airlift.units.Duration;
2021
import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer;
2122
import io.trino.plugin.hudi.util.FileOperationUtils.FileOperation;
2223
import io.trino.testing.AbstractTestQueryFramework;
@@ -36,15 +37,16 @@
3637
import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
3738
import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.getCacheOperationSpans;
3839
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_FG_PT_V8_MOR;
39-
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.DATA;
4040
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.INDEX_DEFINITION;
4141
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.LOG;
4242
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE;
4343
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE_PROPERTIES;
4444
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPERTIES;
4545
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TIMELINE;
4646
import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
47+
import static io.trino.testing.assertions.Assert.assertEventually;
4748
import static java.util.stream.Collectors.toCollection;
49+
import static org.assertj.core.api.Assertions.assertThat;
4850

4951
@ResourceLock("HUDI_CACHE_SYSTEM")
5052
@Execution(ExecutionMode.SAME_THREAD)
@@ -66,10 +68,11 @@ protected DistributedQueryRunner createQueryRunner()
6668
.put("fs.cache.directories", cacheDirectory.toAbsolutePath().toString())
6769
.put("fs.cache.max-sizes", "100MB")
6870
.put("hudi.metadata.cache.enabled", "false")
69-
// Disable async table-statistics refresh: it reads the metadata table on a
70-
// background executor whose spans can outlive the query and leak into the next
71-
// test's measurement (the symmetric off-by-N flake). Disabling it makes the
72-
// file-operation counts deterministic right after the query returns.
71+
// Disable the async table-statistics refresh: on the first query it reads the index
72+
// definitions and table-property files (and the metadata table) on a background
73+
// executor. Those non-metadata-table reads land in the asserted set and their timing
74+
// is non-deterministic, so we turn the refresh off and assert only the synchronous
75+
// planning-path reads.
7376
.put("hudi.table-statistics-enabled", "false")
7477
.buildOrThrow();
7578

@@ -87,12 +90,6 @@ public void testSelectWithFilter()
8790
assertFileSystemAccesses(
8891
query,
8992
ImmutableMultiset.<FileOperation>builder()
90-
.addCopies(new FileOperation("Alluxio.readCached", DATA), 2)
91-
.addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 27)
92-
.addCopies(new FileOperation("Alluxio.readCached", TIMELINE), 4)
93-
.addCopies(new FileOperation("Alluxio.readCached", LOG), 15)
94-
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
95-
.addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 10)
9693
.addCopies(new FileOperation("InputFile.length", TIMELINE), 2)
9794
.addCopies(new FileOperation("InputFile.length", LOG), 1)
9895
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
@@ -103,12 +100,6 @@ public void testSelectWithFilter()
103100
assertFileSystemAccesses(
104101
query,
105102
ImmutableMultiset.<FileOperation>builder()
106-
.addCopies(new FileOperation("Alluxio.readCached", DATA), 2)
107-
.addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 27)
108-
.addCopies(new FileOperation("Alluxio.readCached", TIMELINE), 4)
109-
.addCopies(new FileOperation("Alluxio.readCached", LOG), 15)
110-
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
111-
.addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 10)
112103
.addCopies(new FileOperation("InputFile.length", TIMELINE), 2)
113104
.addCopies(new FileOperation("InputFile.length", LOG), 1)
114105
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
@@ -127,12 +118,6 @@ public void testJoin()
127118

128119
assertFileSystemAccesses(query,
129120
ImmutableMultiset.<FileOperation>builder()
130-
.addCopies(new FileOperation("Alluxio.readCached", DATA), 6)
131-
.addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 215)
132-
.addCopies(new FileOperation("Alluxio.readCached", TIMELINE), 8)
133-
.addCopies(new FileOperation("Alluxio.readCached", LOG), 30)
134-
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 29)
135-
.addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 69)
136121
.addCopies(new FileOperation("InputFile.length", TIMELINE), 4)
137122
.addCopies(new FileOperation("InputFile.length", LOG), 2)
138123
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
@@ -142,12 +127,6 @@ public void testJoin()
142127

143128
assertFileSystemAccesses(query,
144129
ImmutableMultiset.<FileOperation>builder()
145-
.addCopies(new FileOperation("Alluxio.readCached", DATA), 6)
146-
.addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 215)
147-
.addCopies(new FileOperation("Alluxio.readCached", TIMELINE), 8)
148-
.addCopies(new FileOperation("Alluxio.readCached", LOG), 30)
149-
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 29)
150-
.addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 69)
151130
.addCopies(new FileOperation("InputFile.length", TIMELINE), 4)
152131
.addCopies(new FileOperation("InputFile.length", LOG), 2)
153132
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
@@ -156,6 +135,40 @@ public void testJoin()
156135
.build());
157136
}
158137

138+
@Test
139+
public void testReadsServedFromAlluxioCache()
140+
{
141+
// The tests above intentionally do not assert exact Alluxio cache hit/miss counts: the cache
142+
// write is asynchronous, so the per-query counts flake (a write from one query can still be in
143+
// flight when the next query runs). This test instead gives count-independent coverage that the
144+
// Alluxio cache is actually engaged: once the cache is warmed, at least one read is served from
145+
// it (an "Alluxio.readCached" span). assertEventually re-runs the query until the asynchronous
146+
// cache write has landed and a genuine hit is observed, and fails loudly at the deadline if the
147+
// cache never serves a read.
148+
@Language("SQL") String query = "SELECT * FROM " + HUDI_MULTI_FG_PT_V8_MOR;
149+
DistributedQueryRunner queryRunner = getDistributedQueryRunner();
150+
151+
// Warm the cache; the page write into Alluxio happens on a background thread.
152+
queryRunner.executeWithPlan(queryRunner.getDefaultSession(), query);
153+
154+
assertEventually(
155+
Duration.valueOf("30s"),
156+
Duration.valueOf("500ms"),
157+
() -> {
158+
queryRunner.executeWithPlan(queryRunner.getDefaultSession(), query);
159+
assertThat(countCachedReads(queryRunner))
160+
.as("Alluxio.readCached spans (cache hits)")
161+
.isGreaterThanOrEqualTo(1);
162+
});
163+
}
164+
165+
private static long countCachedReads(QueryRunner queryRunner)
166+
{
167+
return getCacheOperationSpans(queryRunner).stream()
168+
.filter(span -> span.getName().equals("Alluxio.readCached"))
169+
.count();
170+
}
171+
159172
private void assertFileSystemAccesses(@Language("SQL") String query, Multiset<FileOperation> expectedCacheAccesses)
160173
{
161174
DistributedQueryRunner queryRunner = getDistributedQueryRunner();
@@ -169,6 +182,14 @@ public static Multiset<FileOperation> getFileOperations(QueryRunner queryRunner)
169182
.stream()
170183
.filter(span -> !span.getName().startsWith("InputFile.exists"))
171184
.map(FileOperation::create)
185+
// Metadata-table reads are issued from Hudi background pools (split loading, partition
186+
// listing, table-statistics refresh) whose spans can outlive the synchronous query and
187+
// land in the next query's measurement window, so their per-query counts are not
188+
// deterministic. Alluxio cache hits/misses (Alluxio.*) depend on whether an earlier
189+
// asynchronous cache write had already completed, so their counts are not deterministic
190+
// either. Both are excluded; only synchronous foreground reads are asserted.
191+
.filter(operation -> operation.fileType() != METADATA_TABLE)
192+
.filter(operation -> !operation.operationType().startsWith("Alluxio."))
172193
.collect(toCollection(HashMultiset::create));
173194
}
174195
}

‎hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java‎

Lines changed: 10 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -56,10 +56,11 @@ protected DistributedQueryRunner createQueryRunner()
5656
.put("hudi.metadata-enabled", "true")
5757
.put("hudi.metadata.cache.enabled", "true")
5858
.put("fs.cache.enabled", "false")
59-
// Disable async table-statistics refresh: it reads the metadata table on a
60-
// background executor whose spans can outlive the query and leak into the next
61-
// test's measurement (the symmetric off-by-N flake). Disabling it makes the
62-
// file-operation counts deterministic right after the query returns.
59+
// Disable the async table-statistics refresh: on the first query it reads the index
60+
// definitions and table-property files (and the metadata table) on a background
61+
// executor. Those non-metadata-table reads land in the asserted set and their timing
62+
// is non-deterministic, so we turn the refresh off and assert only the synchronous
63+
// planning-path reads.
6364
.put("hudi.table-statistics-enabled", "false")
6465
.buildOrThrow();
6566

@@ -78,11 +79,8 @@ public void testSelectWithFilter()
7879
query,
7980
ImmutableMultiset.<FileOperation>builder()
8081
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 2)
81-
.addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 4)
82-
.addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 6)
8382
.addCopies(new FileOperation("FileSystemCache.cacheStream", TIMELINE), 2)
8483
.addCopies(new FileOperation("FileSystemCache.cacheStream", LOG), 1)
85-
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
8684
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
8785
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
8886
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
@@ -92,11 +90,8 @@ public void testSelectWithFilter()
9290
query,
9391
ImmutableMultiset.<FileOperation>builder()
9492
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 2)
95-
.addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 4)
96-
.addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 6)
9793
.addCopies(new FileOperation("FileSystemCache.cacheStream", TIMELINE), 2)
9894
.addCopies(new FileOperation("FileSystemCache.cacheStream", LOG), 1)
99-
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
10095
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
10196
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
10297
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
@@ -114,11 +109,8 @@ public void testJoin()
114109
assertFileSystemAccesses(query,
115110
ImmutableMultiset.<FileOperation>builder()
116111
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 6)
117-
.addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 29)
118-
.addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 40)
119112
.addCopies(new FileOperation("FileSystemCache.cacheStream", TIMELINE), 4)
120113
.addCopies(new FileOperation("FileSystemCache.cacheStream", LOG), 2)
121-
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 29)
122114
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
123115
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
124116
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
@@ -127,11 +119,8 @@ public void testJoin()
127119
assertFileSystemAccesses(query,
128120
ImmutableMultiset.<FileOperation>builder()
129121
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 6)
130-
.addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 29)
131-
.addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 40)
132122
.addCopies(new FileOperation("FileSystemCache.cacheStream", TIMELINE), 4)
133123
.addCopies(new FileOperation("FileSystemCache.cacheStream", LOG), 2)
134-
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 29)
135124
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
136125
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
137126
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
@@ -153,6 +142,11 @@ private static Multiset<FileOperation> getFileOperations(QueryRunner queryRunner
153142
.filter(span -> !span.getName().startsWith("InputFile.exists"))
154143
.filter(span -> !isTrinoSchemaOrPermissions(getFileLocation(span)))
155144
.map(FileOperation::create)
145+
// Metadata-table reads are issued from Hudi background pools (split loading, partition
146+
// listing, table-statistics refresh) whose spans can outlive the synchronous query and
147+
// land in the next query's measurement window. Their per-query counts are therefore
148+
// non-deterministic, so they are excluded; only synchronous foreground reads are asserted.
149+
.filter(operation -> operation.fileType() != METADATA_TABLE)
156150
.collect(toCollection(HashMultiset::create));
157151
}
158152
}

0 commit comments

Comments
 (0)