Re: [PR] fix(metadata): indexing action initializes the partition it asked for [hudi]

["hudi-agent (via GitHub)" <gi...@apache.org>]

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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org