Re: [PR] fix(utilities): reject a missing ordering value at streamer record creation [hudi]

["voonhous (via GitHub)" <gi...@apache.org>]

voonhous commented on code in PR #20049:
URL: https://github.com/apache/hudi/pull/20049#discussion_r4166270379


##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerUtils.java:
##########
@@ -238,4 +250,145 @@ void testCombinePropertiesWithPropsOverride() {
     // propsOverride takes precedence
     assertEquals("overrideValue", result.getString("hoodie.overridden.key"));
   }
+
+  /**
+   * A null value in the ordering field must be quarantined as a record-creation failure rather than
+   * flowing into the write.
+   */
+  @Test
+  void testCreateHoodieRecordsWithNullOrderingValue() {
+    HoodieSchema schema = HoodieSchema.parse(NULLABLE_ORDERING_SCHEMA_STRING);
+    JavaRDD<GenericRecord> recordRdd = jsc.parallelize(Collections.singletonList(1)).map(i -> {
+      GenericRecord genericRecord = new GenericData.Record(schema.toAvroSchema());
+      genericRecord.put(0, i * 1000L);
+      genericRecord.put(1, "key" + i);
+      genericRecord.put(2, "path" + i);
+      genericRecord.put(3, "rider1");
+      genericRecord.put(4, "driver1");
+      genericRecord.put(5, null);
+      return genericRecord;
+    });
+    HoodieStreamer.Config cfg = new HoodieStreamer.Config();
+    cfg.payloadClassName = DefaultHoodieRecordPayload.class.getName();
+    cfg.sourceOrderingFields = ORDERING_FIELD;
+    TypedProperties props = new TypedProperties();
+    props.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), "partition_path");
+    props.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "_row_key");
+    BaseErrorTableWriter errorTableWriter = Mockito.mock(BaseErrorTableWriter.class);
+    ArgumentCaptor<JavaRDD<?>> errorEventCaptor = ArgumentCaptor.forClass(JavaRDD.class);
+    doNothing().when(errorTableWriter).addErrorEvents(errorEventCaptor.capture());
+
+    Option<JavaRDD<HoodieRecord>> recordOpt = HoodieStreamerUtils.createHoodieRecords(
+        cfg, props, Option.of(recordRdd), new SimpleSchemaProvider(jsc, schema, props),
+        HoodieRecordType.AVRO, false, "000", Option.of(errorTableWriter), new HoodieTableConfig());
+
+    assertTrue(recordOpt.isPresent());
+    assertEquals(Collections.emptyList(), recordOpt.get().collect());
+    List<ErrorEvent<String>> errorEvents = (List<ErrorEvent<String>>) errorEventCaptor.getValue().collect();
+    // The record is schema-valid, so it is serialized by the Avro JsonEncoder, which wraps
+    // nullable union values.
+    ErrorEvent<String> expectedErrorEvent = new ErrorEvent<>(
+        "{\"timestamp\":1000,\"_row_key\":\"key1\",\"partition_path\":{\"string\":\"path1\"},"
+            + "\"rider\":\"rider1\",\"driver\":\"driver1\",\"" + ORDERING_FIELD + "\":null}",
+        ErrorEvent.ErrorReason.RECORD_CREATION);
+    assertEquals(Collections.singletonList(expectedErrorEvent), errorEvents);
+  }
+
+  /**
+   * With several ordering fields, a null in any one of them is still a missing ordering value:
+   * OrderingValues.create returns a non-null ArrayComparable holding the null, so it survives a
+   * plain null check and fails later when the merger compares it.
+   */
+  @Test
+  void testCreateHoodieRecordsWithNullInOneOfSeveralOrderingFields() {
+    HoodieSchema schema = HoodieSchema.parse(NULLABLE_ORDERING_SCHEMA_STRING);
+    JavaRDD<GenericRecord> recordRdd = nullOrderingRecords(schema, false);
+    HoodieStreamer.Config cfg = nullOrderingConfig("timestamp," + ORDERING_FIELD);
+
+    List<ErrorEvent<String>> errorEvents = quarantinedEvents(cfg, schema, recordRdd);
+
+    assertEquals(1, errorEvents.size());
+    assertEquals(ErrorEvent.ErrorReason.RECORD_CREATION, errorEvents.get(0).getReason());
+  }
+
+  /**
+   * A delete carrying a null ordering value is rejected too. It is not a commit-time-ordering
+   * delete, since that requires the default ordering value rather than a null one, so the merger
+   * would compare the null and fail.
+   */
+  @Test
+  void testCreateHoodieRecordsWithNullOrderingValueOnDelete() {
+    HoodieSchema schema = HoodieSchema.parse(NULLABLE_ORDERING_DELETE_SCHEMA_STRING);
+    JavaRDD<GenericRecord> recordRdd = nullOrderingRecords(schema, true);
+    HoodieStreamer.Config cfg = nullOrderingConfig(ORDERING_FIELD);
+
+    List<ErrorEvent<String>> errorEvents = quarantinedEvents(cfg, schema, recordRdd);
+
+    assertEquals(1, errorEvents.size());
+    assertEquals(ErrorEvent.ErrorReason.RECORD_CREATION, errorEvents.get(0).getReason());
+  }
+
+  /**
+   * Without an error table there is nowhere to quarantine the record, so the batch fails instead of
+   * writing an unusable ordering value.
+   */
+  @Test
+  void testCreateHoodieRecordsWithNullOrderingValueFailsWithoutErrorTable() {
+    HoodieSchema schema = HoodieSchema.parse(NULLABLE_ORDERING_SCHEMA_STRING);
+    JavaRDD<GenericRecord> recordRdd = nullOrderingRecords(schema, false);
+    HoodieStreamer.Config cfg = nullOrderingConfig(ORDERING_FIELD);
+
+    Option<JavaRDD<HoodieRecord>> recordOpt = HoodieStreamerUtils.createHoodieRecords(
+        cfg, nullOrderingProps(), Option.of(recordRdd), new SimpleSchemaProvider(jsc, schema, nullOrderingProps()),
+        HoodieRecordType.AVRO, false, "000", Option.empty(), new HoodieTableConfig());
+
+    assertTrue(recordOpt.isPresent());
+    SparkException sparkException = assertThrows(SparkException.class, () -> recordOpt.get().collect());
+    assertEquals(HoodieRecordCreationException.class, sparkException.getCause().getClass());
+  }
+
+  private static TypedProperties nullOrderingProps() {
+    TypedProperties props = new TypedProperties();
+    props.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), "partition_path");
+    props.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "_row_key");
+    return props;
+  }
+
+  private static HoodieStreamer.Config nullOrderingConfig(String orderingFields) {
+    HoodieStreamer.Config cfg = new HoodieStreamer.Config();
+    cfg.payloadClassName = DefaultHoodieRecordPayload.class.getName();
+    cfg.sourceOrderingFields = orderingFields;
+    return cfg;
+  }

Review Comment:
   All four new tests use this function, so `requiresOrderingValue` is always true. No test reaches the false gate. This config causes the `cfg.recordMergeMode` to be pinned to null, which IIUC, `HoodieStreamer` never produces, and passes an empty `HoodieTableConfig`.
   
   Could we adde a parameterized case for `COMMIT_TIME_ORDERING` and `OverwriteWithLatestAvroPayload` asserting the null-ordering record is returned and nothing is quarantined? 



##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamerUtils.java:
##########
@@ -97,6 +99,9 @@ public static Option<JavaRDD<HoodieRecord>> createHoodieRecords(HoodieStreamer.C
     String payloadClassName = StringUtils.isNullOrEmpty(cfg.payloadClassName)
         ? HoodieRecordPayload.getAvroPayloadForMergeMode(cfg.recordMergeMode, cfg.payloadClassName)
         : cfg.payloadClassName;
+    boolean requiresOrderingValue = shouldUseOrderingField
+        && cfg.recordMergeMode != RecordMergeMode.COMMIT_TIME_ORDERING

Review Comment:
   +1 on this, please chec.



##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamerUtils.java:
##########
@@ -130,6 +135,13 @@ public static Option<JavaRDD<HoodieRecord>> createHoodieRecords(HoodieStreamer.C
                       ? OrderingValues.create(orderingFieldsStr.split(","),
                          field -> (Comparable) HoodieAvroUtils.getNestedFieldVal(gr, field, false, useConsistentLogicalTimestamp))
                       : null;
+                  if (requiresOrderingValue && OrderingValues.isMissing(orderingValue)) {
+                    throw new IllegalArgumentException(
+                        "Ordering fields '" + orderingFieldsStr + "' resolved to a null value for record key '"
+                            + hoodieKey.getRecordKey() + "'. Please ensure all records carry non-null values for "
+                            + "the ordering fields, or use a merge mode or payload class that does not order "
+                            + "(e.g., COMMIT_TIME_ORDERING or OverwriteWithLatestAvroPayload).");

Review Comment:
   IIUC, this impacts MOR deletes. MOR deletes with a null ordering value, which this PR will quarantine them. 
   
   Could deletes skip the check and carry `OrderingValues.getDefault()`, or should the Summary of the PR be corrected and an impact line for MOR is added?



##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerUtils.java:
##########
@@ -238,4 +250,145 @@ void testCombinePropertiesWithPropsOverride() {
     // propsOverride takes precedence
     assertEquals("overrideValue", result.getString("hoodie.overridden.key"));
   }
+
+  /**
+   * A null value in the ordering field must be quarantined as a record-creation failure rather than
+   * flowing into the write.
+   */
+  @Test
+  void testCreateHoodieRecordsWithNullOrderingValue() {
+    HoodieSchema schema = HoodieSchema.parse(NULLABLE_ORDERING_SCHEMA_STRING);
+    JavaRDD<GenericRecord> recordRdd = jsc.parallelize(Collections.singletonList(1)).map(i -> {
+      GenericRecord genericRecord = new GenericData.Record(schema.toAvroSchema());
+      genericRecord.put(0, i * 1000L);
+      genericRecord.put(1, "key" + i);
+      genericRecord.put(2, "path" + i);
+      genericRecord.put(3, "rider1");
+      genericRecord.put(4, "driver1");
+      genericRecord.put(5, null);
+      return genericRecord;
+    });
+    HoodieStreamer.Config cfg = new HoodieStreamer.Config();
+    cfg.payloadClassName = DefaultHoodieRecordPayload.class.getName();
+    cfg.sourceOrderingFields = ORDERING_FIELD;
+    TypedProperties props = new TypedProperties();
+    props.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), "partition_path");
+    props.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "_row_key");
+    BaseErrorTableWriter errorTableWriter = Mockito.mock(BaseErrorTableWriter.class);
+    ArgumentCaptor<JavaRDD<?>> errorEventCaptor = ArgumentCaptor.forClass(JavaRDD.class);
+    doNothing().when(errorTableWriter).addErrorEvents(errorEventCaptor.capture());
+
+    Option<JavaRDD<HoodieRecord>> recordOpt = HoodieStreamerUtils.createHoodieRecords(
+        cfg, props, Option.of(recordRdd), new SimpleSchemaProvider(jsc, schema, props),
+        HoodieRecordType.AVRO, false, "000", Option.of(errorTableWriter), new HoodieTableConfig());
+
+    assertTrue(recordOpt.isPresent());
+    assertEquals(Collections.emptyList(), recordOpt.get().collect());
+    List<ErrorEvent<String>> errorEvents = (List<ErrorEvent<String>>) errorEventCaptor.getValue().collect();
+    // The record is schema-valid, so it is serialized by the Avro JsonEncoder, which wraps
+    // nullable union values.
+    ErrorEvent<String> expectedErrorEvent = new ErrorEvent<>(
+        "{\"timestamp\":1000,\"_row_key\":\"key1\",\"partition_path\":{\"string\":\"path1\"},"
+            + "\"rider\":\"rider1\",\"driver\":\"driver1\",\"" + ORDERING_FIELD + "\":null}",
+        ErrorEvent.ErrorReason.RECORD_CREATION);
+    assertEquals(Collections.singletonList(expectedErrorEvent), errorEvents);
+  }
+
+  /**
+   * With several ordering fields, a null in any one of them is still a missing ordering value:
+   * OrderingValues.create returns a non-null ArrayComparable holding the null, so it survives a
+   * plain null check and fails later when the merger compares it.
+   */
+  @Test
+  void testCreateHoodieRecordsWithNullInOneOfSeveralOrderingFields() {
+    HoodieSchema schema = HoodieSchema.parse(NULLABLE_ORDERING_SCHEMA_STRING);
+    JavaRDD<GenericRecord> recordRdd = nullOrderingRecords(schema, false);
+    HoodieStreamer.Config cfg = nullOrderingConfig("timestamp," + ORDERING_FIELD);
+
+    List<ErrorEvent<String>> errorEvents = quarantinedEvents(cfg, schema, recordRdd);
+
+    assertEquals(1, errorEvents.size());
+    assertEquals(ErrorEvent.ErrorReason.RECORD_CREATION, errorEvents.get(0).getReason());
+  }
+
+  /**
+   * A delete carrying a null ordering value is rejected too. It is not a commit-time-ordering
+   * delete, since that requires the default ordering value rather than a null one, so the merger
+   * would compare the null and fail.
+   */
+  @Test
+  void testCreateHoodieRecordsWithNullOrderingValueOnDelete() {
+    HoodieSchema schema = HoodieSchema.parse(NULLABLE_ORDERING_DELETE_SCHEMA_STRING);
+    JavaRDD<GenericRecord> recordRdd = nullOrderingRecords(schema, true);
+    HoodieStreamer.Config cfg = nullOrderingConfig(ORDERING_FIELD);
+
+    List<ErrorEvent<String>> errorEvents = quarantinedEvents(cfg, schema, recordRdd);
+
+    assertEquals(1, errorEvents.size());
+    assertEquals(ErrorEvent.ErrorReason.RECORD_CREATION, errorEvents.get(0).getReason());
+  }
+
+  /**
+   * Without an error table there is nowhere to quarantine the record, so the batch fails instead of
+   * writing an unusable ordering value.
+   */
+  @Test
+  void testCreateHoodieRecordsWithNullOrderingValueFailsWithoutErrorTable() {
+    HoodieSchema schema = HoodieSchema.parse(NULLABLE_ORDERING_SCHEMA_STRING);
+    JavaRDD<GenericRecord> recordRdd = nullOrderingRecords(schema, false);
+    HoodieStreamer.Config cfg = nullOrderingConfig(ORDERING_FIELD);
+
+    Option<JavaRDD<HoodieRecord>> recordOpt = HoodieStreamerUtils.createHoodieRecords(
+        cfg, nullOrderingProps(), Option.of(recordRdd), new SimpleSchemaProvider(jsc, schema, nullOrderingProps()),
+        HoodieRecordType.AVRO, false, "000", Option.empty(), new HoodieTableConfig());
+
+    assertTrue(recordOpt.isPresent());
+    SparkException sparkException = assertThrows(SparkException.class, () -> recordOpt.get().collect());
+    assertEquals(HoodieRecordCreationException.class, sparkException.getCause().getClass());

Review Comment:
   no-error-table test asserts only the `HoodieRecordCreationException` wrapper, which any creation failure  produces. There are multi field delete tests above that assert only `RECORD_CREATION` and `ErrorEvent` carries no throwable.
   
   Can we parameterize the test over three shapes and assert the `IllegalArgumentException` message instead?



-- 
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