hudi-agent commented on code in PR #19717:
URL: https://github.com/apache/hudi/pull/19717#discussion_r3900969689
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/BaseIndexer.java:
##########
@@ -51,6 +56,54 @@ protected BaseIndexer(
this.dataTableMetaClient = dataTableMetaClient;
}
+ /**
+ * Resolves which partition of a definition-driven index type to initialize.
+ * <p>
+ * An indexing action names the partition in the context, and that partition
is initialized
+ * whenever its index definition exists, regardless of how many other
definitions of the type
+ * are still uninitialized. A regular write names nothing, and the partition
is inferred from
+ * the uninitialized definitions: exactly one means that one, any other
count means nothing
+ * is initialized. A requested partition without a definition goes through
the same inference
+ * (that is where a first-time index mints its definition from the write
config), but when the
+ * inference cannot resolve to exactly one partition the action fails rather
than completing
+ * with nothing built and the requested partition marked complete.
+ *
+ * @param context the initialization context
+ * @param uninitializedPartitions the uninitialized partitions of this
type, as the definition
+ * lookup reports them
+ * @param indexType the index type, for the messages
+ * @return the partitions to initialize: exactly one, or none
+ */
+ protected Set<String> resolvePartitionsToInit(IndexInitializationContext
context,
+ Set<String>
uninitializedPartitions,
+ MetadataPartitionType
indexType) {
+ Option<String> requested = context.requestedIndexPartition();
+ if (requested.isPresent()
+ &&
dataTableMetaClient.getTableConfig().getMetadataPartitions().contains(requested.get()))
{
+ // Already initialized, typically by the write that committed between
scheduling and running the
+ // indexing action. Re-initializing would commit at an instant the
metadata table already holds
+ // completed, and the rollback-and-recommit inside that commit destroys
the earlier commit's records.
+ log.info("Metadata partition {} is already initialized, skipping",
requested.get());
+ return Collections.emptySet();
+ }
+ if (requested.isPresent() &&
dataTableMetaClient.getIndexForMetadataPartition(requested.get()).isPresent()) {
+ return Collections.singleton(requested.get());
+ }
+ if (uninitializedPartitions.size() == 1) {
Review Comment:
🤖 If the requested partition has no index definition yet, the branch above
is skipped and we land here. But `uninitializedPartitions` only ever contains
partitions that already have definitions, so when exactly one other same-type
index is uninitialized this returns that *other* partition — building the wrong
index while the plan still marks the requested one inflight/complete. Could
this be hit by a HoodieIndexer bootstrap of a new SI/EI while another same-type
definition sits unbuilt, and if so should the requested-without-definition case
fall through to the throw instead?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieIndexer.java:
##########
@@ -181,6 +187,92 @@ public void testIndexerForRecordIndex() {
"streamer-config/indexer-record-index.properties");
}
+ /**
+ * An indexing action builds the partition it names. With two
secondary-index and two expression-index definitions
+ * registered and none initialized (the shape of a table whose metadata
table was rebuilt with its definitions
+ * intact), each action builds exactly the index it names, on storage and
not only in the table config, until all
+ * four are built.
+ */
+ @Test
+ void testIndexerBuildsEachRequestedIndexAmongSeveralUninitialized() {
+ String tableName = "indexer_test_two_si_two_ei";
+ HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder()
+
.enable(true).withAsyncIndex(false).withMetadataIndexColumnStats(false).build();
+ upsertToTable(metadataConfig, tableName);
+ indexMetadataPartitionsAndAssert(RECORD_INDEX.getPartitionPath(),
Collections.singletonList(FILES), Arrays.asList(COLUMN_STATS, BLOOM_FILTERS),
tableName,
+ "streamer-config/indexer-record-index.properties");
+
+ metaClient = reload(metaClient);
+ String riderSecondaryIndex = SECONDARY_INDEX.getPartitionPath() +
"idx_rider";
+ String driverSecondaryIndex = SECONDARY_INDEX.getPartitionPath() +
"idx_driver";
+ String riderExpressionIndex = EXPRESSION_INDEX.getPartitionPath() +
"idx_rider_expr";
+ String driverExpressionIndex = EXPRESSION_INDEX.getPartitionPath() +
"idx_driver_expr";
+
metaClient.buildIndexDefinition(secondaryIndexDefinition(riderSecondaryIndex,
"rider"));
+
metaClient.buildIndexDefinition(secondaryIndexDefinition(driverSecondaryIndex,
"driver"));
+
metaClient.buildIndexDefinition(expressionIndexDefinition(riderExpressionIndex,
"rider"));
+
metaClient.buildIndexDefinition(expressionIndexDefinition(driverExpressionIndex,
"driver"));
+
+ indexMetadataPartitionsAndAssert(riderSecondaryIndex, Arrays.asList(FILES,
RECORD_INDEX), Arrays.asList(COLUMN_STATS, BLOOM_FILTERS), tableName,
+ "streamer-config/indexer-secondary-index.properties");
+ assertIndexesNotBuilt(Arrays.asList(driverSecondaryIndex,
riderExpressionIndex, driverExpressionIndex));
+
+ indexMetadataPartitionsAndAssert(driverSecondaryIndex,
Arrays.asList(FILES, RECORD_INDEX), Arrays.asList(COLUMN_STATS, BLOOM_FILTERS),
tableName,
+ "streamer-config/indexer-secondary-index.properties",
+ Arrays.asList("hoodie.index.name=idx_driver",
"hoodie.metadata.index.secondary.column=driver"));
+ assertIndexesNotBuilt(Arrays.asList(riderExpressionIndex,
driverExpressionIndex));
+
+ indexMetadataPartitionsAndAssert(riderExpressionIndex,
Arrays.asList(FILES, RECORD_INDEX), Arrays.asList(COLUMN_STATS, BLOOM_FILTERS),
tableName,
+ "streamer-config/indexer-expression-index.properties",
+ Arrays.asList("hoodie.index.name=idx_rider_expr",
"hoodie.metadata.index.expression.column=rider"));
+ assertIndexesNotBuilt(Collections.singletonList(driverExpressionIndex));
+
+ indexMetadataPartitionsAndAssert(driverExpressionIndex,
Arrays.asList(FILES, RECORD_INDEX), Arrays.asList(COLUMN_STATS, BLOOM_FILTERS),
tableName,
+ "streamer-config/indexer-expression-index.properties",
+ Arrays.asList("hoodie.index.name=idx_driver_expr",
"hoodie.metadata.index.expression.column=driver"));
+
+ assertIndexesBuiltWithFileSlices(Arrays.asList(riderSecondaryIndex,
driverSecondaryIndex, riderExpressionIndex, driverExpressionIndex));
+ }
+
+ private HoodieIndexDefinition secondaryIndexDefinition(String fullIndexName,
String sourceField) {
+ return HoodieIndexDefinition.newBuilder()
+ .withIndexName(fullIndexName)
+ .withIndexType(PARTITION_NAME_SECONDARY_INDEX)
+ .withIndexFunction(IDENTITY_TRANSFORM)
+ .withSourceFields(Collections.singletonList(sourceField))
+
.withVersion(HoodieIndexVersion.getCurrentVersion(metaClient.getTableConfig().getTableVersion(),
SECONDARY_INDEX))
+ .build();
+ }
+
+ private HoodieIndexDefinition expressionIndexDefinition(String
fullIndexName, String sourceField) {
+ return HoodieIndexDefinition.newBuilder()
+ .withIndexName(fullIndexName)
+ .withIndexType(PARTITION_NAME_COLUMN_STATS)
+ .withIndexFunction(IDENTITY_TRANSFORM)
+ .withSourceFields(Collections.singletonList(sourceField))
+
.withVersion(HoodieIndexVersion.getCurrentVersion(metaClient.getTableConfig().getTableVersion(),
EXPRESSION_INDEX))
+ .build();
+ }
+
+ private void assertIndexesNotBuilt(List<String> indexPartitions) {
+ metaClient = reload(metaClient);
+ for (String indexPartition : indexPartitions) {
+
assertFalse(metaClient.getTableConfig().getMetadataPartitions().contains(indexPartition),
Review Comment:
🤖 nit: this helper is named expressionIndexDefinition but builds with
withIndexType(PARTITION_NAME_COLUMN_STATS) — is that intentional? It reads as a
copy-paste and will confuse a future reader; if it's deliberate a one-line
comment would help.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/secondary/TestSecondaryIndexer.java:
##########
@@ -71,6 +73,84 @@ void testSkipWhenMultipleSecondaryPartitions() throws
IOException {
}
}
+ @Test
+ void testInitializesTheRequestedPartitionAmongSeveralUninitialized() throws
IOException {
+ HoodieEngineContext engineContext = new
HoodieLocalEngineContext(getDefaultStorageConf());
+ HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+ HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ HoodieIndexDefinition definition = mock(HoodieIndexDefinition.class);
+
+ when(writeConfig.getMetadataConfig()).thenReturn(metadataConfig);
+ when(metadataConfig.getSecondaryIndexParallelism()).thenReturn(8);
+ when(writeConfig.getProps()).thenReturn(new TypedProperties());
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+
when(tableConfig.getMetadataPartitions()).thenReturn(Collections.emptySet());
Review Comment:
🤖 nit: could you add an import for SecondaryIndexRecordGenerationUtils
rather than spelling out the fully-qualified name inline in both the
MockedStatic type and the when(...) call? It's quite noisy to read here.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
