shangxinli commented on code in PR #18765: URL: https://github.com/apache/hudi/pull/18765#discussion_r3277721910
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SuccessfulRecordCounter.java: ########## @@ -0,0 +1,111 @@ +/* + * 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.utilities.streamer; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.util.Option; + +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.function.Function; + +import java.util.List; +import java.util.Objects; + +/** + * Computes record counts for a HoodieStreamer commit, summing across the data-table + * write statuses and (optionally) the error-table write statuses when error-table + * write unification is enabled. + * + * <p>Extracted from {@code HoodieStreamerWriteStatusValidator} (issue #18750) so the + * counting logic can be invoked from the explicit pre-commit orchestration in + * {@code StreamSync} without going through the {@code WriteStatusValidator} callback.</p> + */ +public final class SuccessfulRecordCounter { + + private SuccessfulRecordCounter() { + } + + /** + * Compute total / errored / successful record counts from a pre-collected list of write statuses. + * + * @param dataTableWriteStatuses Pre-collected data-table write statuses. Must not be null. + * @param errorTableWriteStatusRDDOpt Optional error-table write status RDD; only consulted + * when unification is enabled. Must not be null + * ({@link Option#empty()} when no error table). + * @param isErrorTableWriteUnificationEnabled Whether error-table records contribute to the totals. + * @return immutable {@link Counts} snapshot. + */ + public static Counts compute(List<WriteStatus> dataTableWriteStatuses, + Option<JavaRDD<WriteStatus>> errorTableWriteStatusRDDOpt, + boolean isErrorTableWriteUnificationEnabled) { + Objects.requireNonNull(dataTableWriteStatuses, "dataTableWriteStatuses"); + Objects.requireNonNull(errorTableWriteStatusRDDOpt, "errorTableWriteStatusRDDOpt"); + + long totalRecords = 0L; + long totalErroredRecords = 0L; + for (WriteStatus ws : dataTableWriteStatuses) { + totalRecords += ws.getTotalRecords(); + totalErroredRecords += ws.getTotalErrorRecords(); + } + if (isErrorTableWriteUnificationEnabled && errorTableWriteStatusRDDOpt.isPresent()) { + JavaRDD<WriteStatus> errorRdd = errorTableWriteStatusRDDOpt.get(); + totalRecords += sumLong(errorRdd, WriteStatus::getTotalRecords); + totalErroredRecords += sumLong(errorRdd, WriteStatus::getTotalErrorRecords); + } + return new Counts(totalRecords, totalErroredRecords); + } + + /** + * Lossless long sum over an RDD. Avoids the precision loss of {@code mapToDouble().sum()}, + * which silently rounds counts above 2^53 (about 9 quadrillion). + */ + private static long sumLong(JavaRDD<WriteStatus> rdd, Function<WriteStatus, Long> extractor) { + return rdd.map(extractor).fold(0L, Long::sum); + } + + /** Immutable count snapshot. */ + public static final class Counts { + public static final Counts ZERO = new Counts(0L, 0L); + + private final long totalRecords; + private final long totalErroredRecords; + + public Counts(long totalRecords, long totalErroredRecords) { + this.totalRecords = totalRecords; + this.totalErroredRecords = totalErroredRecords; + } + + public long getTotalRecords() { + return totalRecords; + } + + public long getTotalErroredRecords() { Review Comment: Done in 140e17f. Renamed `getTotalErroredRecords()` / `totalErroredRecords` → `getTotalErrorRecords()` / `totalErrorRecords` on `SuccessfulRecordCounter.Counts` (constructor, field, getter, internal locals) and propagated the rename through `StreamSync` callers, the `SparkWriteErrorValidator` Javadoc, and `TestSuccessfulRecordCounter`. Terminology now matches `WriteStatus.getTotalErrorRecords()`. -- 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]
