vinothchandar commented on code in PR #19212:
URL: https://github.com/apache/hudi/pull/19212#discussion_r3567840655
##########
hudi-common/src/main/java/org/apache/hudi/common/avro/HoodieAvroUtils.java:
##########
@@ -1407,17 +1412,87 @@ public static HoodieRecord createHoodieRecordFromAvro(
boolean populateMetaFields,
Option<HoodieSchema> schemaWithoutMetaFields) {
if (populateMetaFields) {
- return SpillableMapUtils.convertToHoodieRecordPayload((GenericRecord)
data,
+ return convertToHoodieRecordPayload((GenericRecord) data,
payloadClass, preCombineFields, withOperation);
} else if (simpleKeyGenFieldsOpt.isPresent()) {
- return SpillableMapUtils.convertToHoodieRecordPayload((GenericRecord)
data,
+ return convertToHoodieRecordPayload((GenericRecord) data,
payloadClass, preCombineFields, simpleKeyGenFieldsOpt.get(),
withOperation, partitionNameOp, schemaWithoutMetaFields);
} else {
- return SpillableMapUtils.convertToHoodieRecordPayload((GenericRecord)
data,
+ return convertToHoodieRecordPayload((GenericRecord) data,
payloadClass, preCombineFields, withOperation, partitionNameOp,
schemaWithoutMetaFields);
}
}
+ /**
+ * Utility method to convert bytes to HoodieRecord using schema and payload
class.
+ */
+ public static <R> HoodieRecord<R> convertToHoodieRecordPayload(GenericRecord
rec, String payloadClazz, String[] preCombineFields, boolean
withOperationField) {
+ return convertToHoodieRecordPayload(rec, payloadClazz, preCombineFields,
+ Pair.of(HoodieRecord.RECORD_KEY_METADATA_FIELD,
HoodieRecord.PARTITION_PATH_METADATA_FIELD),
+ withOperationField, Option.empty(), Option.empty());
+ }
+
+ public static <R> HoodieRecord<R> convertToHoodieRecordPayload(GenericRecord
record, String payloadClazz,
+ String[]
preCombineFields,
+ boolean
withOperationField,
+
Option<String> partitionName,
+
Option<HoodieSchema> schemaWithoutMetaFields) {
+ return convertToHoodieRecordPayload(record, payloadClazz, preCombineFields,
+ Pair.of(HoodieRecord.RECORD_KEY_METADATA_FIELD,
HoodieRecord.PARTITION_PATH_METADATA_FIELD),
+ withOperationField, partitionName, schemaWithoutMetaFields);
+ }
+
+ /**
+ * Utility method to convert bytes to HoodieRecord using schema and payload
class.
+ */
+ public static <R> HoodieRecord<R> convertToHoodieRecordPayload(GenericRecord
record, String payloadClazz,
+ String[]
preCombineFields,
+ Pair<String,
String> recordKeyPartitionPathFieldPair,
+ boolean
withOperationField,
+
Option<String> partitionName,
+
Option<HoodieSchema> schemaWithoutMetaFields) {
+ final String recKey =
record.get(recordKeyPartitionPathFieldPair.getKey()).toString();
+ final String partitionPath = (partitionName.isPresent() ?
partitionName.get() :
+ record.get(recordKeyPartitionPathFieldPair.getRight()).toString());
+
+ Comparable preCombineVal = getPreCombineVal(record, preCombineFields);
+ HoodieOperation operation = withOperationField
+ ? HoodieOperation.fromName(getNullableValAsString(record,
HoodieRecord.OPERATION_METADATA_FIELD)) : null;
+
+ if (schemaWithoutMetaFields.isPresent()) {
+ Schema schema = schemaWithoutMetaFields.get().toAvroSchema();
+ GenericRecord recordWithoutMetaFields = new GenericData.Record(schema);
+ for (Schema.Field f : schema.getFields()) {
+ recordWithoutMetaFields.put(f.pos(), record.get(f.name()));
+ }
+ record = recordWithoutMetaFields;
+ }
+
+ HoodieRecord<? extends HoodieRecordPayload> hoodieRecord = new
HoodieAvroRecord<>(new HoodieKey(recKey, partitionPath),
+ HoodieRecordUtils.loadPayload(payloadClazz, record, preCombineVal),
operation);
+ return (HoodieRecord<R>) hoodieRecord;
+ }
+
+ /**
+ * Returns the preCombine value with given field name.
+ *
+ * @param rec The avro record
+ * @param preCombineFields The preCombine field names
+ * @return the preCombine field value or 0 if the field does not exist in
the avro schema
+ */
+ private static Comparable getPreCombineVal(GenericRecord rec, @Nullable
String[] preCombineFields) {
Review Comment:
ensure UTs for this. need to add more if needed
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/MergeUtils.java:
##########
@@ -41,7 +41,7 @@
import static
org.apache.hudi.common.config.HoodieMemoryConfig.MAX_MEMORY_FRACTION_FOR_MERGE;
@Slf4j
-public class IOUtils {
+public class MergeUtils {
Review Comment:
are n't there other Merge related Utils to group here?
##########
hudi-io/src/main/java/org/apache/hudi/common/util/ValidationUtils.java:
##########
@@ -81,10 +81,12 @@ public static void checkState(final boolean expression,
String errorMessage) {
/**
* Ensures the truth of an expression, throwing the custom errorMessage
otherwise.
+ *
+ * @throws IllegalStateException if {@code expression} is false
*/
public static void checkState(final boolean expression, final
Supplier<String> errorMessageSupplier) {
if (!expression) {
- throw new IllegalArgumentException(errorMessageSupplier.get());
+ throw new IllegalStateException(errorMessageSupplier.get());
Review Comment:
check once there are no unintended consequences for callers
##########
hudi-common/src/main/java/org/apache/hudi/common/avro/HoodieAvroUtils.java:
##########
@@ -1407,17 +1412,87 @@ public static HoodieRecord createHoodieRecordFromAvro(
boolean populateMetaFields,
Option<HoodieSchema> schemaWithoutMetaFields) {
if (populateMetaFields) {
- return SpillableMapUtils.convertToHoodieRecordPayload((GenericRecord)
data,
+ return convertToHoodieRecordPayload((GenericRecord) data,
payloadClass, preCombineFields, withOperation);
} else if (simpleKeyGenFieldsOpt.isPresent()) {
- return SpillableMapUtils.convertToHoodieRecordPayload((GenericRecord)
data,
+ return convertToHoodieRecordPayload((GenericRecord) data,
payloadClass, preCombineFields, simpleKeyGenFieldsOpt.get(),
withOperation, partitionNameOp, schemaWithoutMetaFields);
} else {
- return SpillableMapUtils.convertToHoodieRecordPayload((GenericRecord)
data,
+ return convertToHoodieRecordPayload((GenericRecord) data,
payloadClass, preCombineFields, withOperation, partitionNameOp,
schemaWithoutMetaFields);
}
}
+ /**
+ * Utility method to convert bytes to HoodieRecord using schema and payload
class.
+ */
+ public static <R> HoodieRecord<R> convertToHoodieRecordPayload(GenericRecord
rec, String payloadClazz, String[] preCombineFields, boolean
withOperationField) {
Review Comment:
rename to `convertToRecord`
##########
hudi-common/src/main/java/org/apache/hudi/common/util/ConfigUtils.java:
##########
@@ -411,6 +411,32 @@ public static Option<Object>
getRawValueWithAltKeys(Properties props,
return Option.empty();
}
+ /**
+ * Gets the raw value for a {@link ConfigProperty} config using a key
mapping function. The key
+ * and alternative keys are used to fetch the config. Unlike
+ * {@link #getStringWithAltKeys(Function, ConfigProperty)}, this does not
fall back to the
+ * config's default value when the config is not found.
+ *
+ * @param keyMapper Mapper function to map the key to values.
+ * @param configProperty {@link ConfigProperty} config to fetch.
+ * @return {@link Option} of value if the config exists; empty {@link
Option} otherwise.
+ */
+ public static Option<Object> getRawValueWithAltKeys(Function<String, Object>
keyMapper,
Review Comment:
confirm this is a method relocation with no other changes.
##########
hudi-spark-datasource/hudi-spark4-common/src/main/java/org/apache/hudi/variant/Spark4VariantShreddingProvider.java:
##########
@@ -250,17 +252,6 @@ private VariantSchema.ScalarType
avroTypeToScalarType(Schema schema) {
}
}
- private static Schema unwrapNullable(Schema schema) {
Review Comment:
revisit specifically to understand union handling at caller site
##########
hudi-common/src/main/java/org/apache/hudi/common/util/FileFormatUtils.java:
##########
@@ -302,10 +308,14 @@ protected HoodieSchema getKeyIteratorSchema(HoodieStorage
storage, StoragePath f
* @param partitionPath optional partition path for the file, if provided
only the record key is read from the file
* @return {@link Iterator} of pairs of {@link HoodieKey} and position
fetched from the data file.
*/
- public abstract ClosableIterator<Pair<HoodieKey, Long>>
fetchRecordKeysWithPositions(HoodieStorage storage,
-
StoragePath filePath,
-
Option<BaseKeyGenerator> keyGeneratorOpt,
-
Option<String> partitionPath);
+ public ClosableIterator<Pair<HoodieKey, Long>>
fetchRecordKeysWithPositions(HoodieStorage storage,
Review Comment:
`HFileUtils` should still throw the exception as before?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/PulsarSource.java:
##########
@@ -264,8 +263,22 @@ private static void shutdownPulsarClient(PulsarClient
client) throws PulsarClien
// properly (see above). To work this around we employ "nuclear"
option of
// fetching all Pulsar client threads and interrupting them forcibly
(to make them
// shutdown)
- collectActiveThreads().stream().sequential()
+ Arrays.stream(collectActiveThreads())
.filter(t -> t.getName().startsWith("pulsar-client-io"))
.forEach(Thread::interrupt);
}
+
+ /**
+ * Fetches all active threads currently running in the JVM.
+ */
+ private static Thread[] collectActiveThreads() {
Review Comment:
add test?
##########
hudi-common/src/main/java/org/apache/hudi/common/util/TablePathUtils.java:
##########
@@ -102,14 +102,16 @@ private static Option<StoragePath>
getTablePathFromPartitionPath(HoodieStorage s
} else {
// Simply traverse directory structure until found .hoodie folder
StoragePath current = partitionPath;
- while (current != null) {
+ while (true) {
if (hasTableMetadataFolder(storage, current)) {
return Option.of(current);
}
+ if (current.depth() == 0) {
Review Comment:
add UT. need to ensure the depth eventually becomes 0 at root.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieVariantReconstruction.java:
##########
@@ -146,8 +146,8 @@ static HoodieVariantReconstruction create(HoodieSchema
fileSchema, HoodieSchema
Schema[] unshreddedSubSchemas = new Schema[requestedFields.size()];
for (int i = 0; i < requestedFields.size(); i++) {
if (isTarget[i]) {
- shreddedSubSchemas[i] =
unwrapNullable(fileSchema.getField(requestedFields.get(i).name()).get().schema()).getAvroSchema();
- unshreddedSubSchemas[i] =
unwrapNullable(outputSchema.getFields().get(i).schema()).getAvroSchema();
+ shreddedSubSchemas[i] =
fileSchema.getField(requestedFields.get(i).name()).get().schema().getNonNullType().getAvroSchema();
Review Comment:
need new UTs to verify the behavior is same
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java:
##########
@@ -381,29 +379,15 @@ public static Map<Path, HoodieTableMetaClient>
getTableMetaClientByPartitionPath
* Extract HoodieTableMetaClient from a partition path (not base path)
*/
public static HoodieTableMetaClient
getTableMetaClientForBasePathUnchecked(Configuration conf, Path partitionPath)
throws IOException {
- Path baseDir = partitionPath;
HoodieStorage storage = HoodieStorageUtils.getStorage(
partitionPath.toString(), HadoopFSUtils.getStorageConf(conf));
- StoragePath partitionStoragePath = convertToStoragePath(partitionPath);
- if (HoodiePartitionMetadata.hasPartitionMetadata(storage,
partitionStoragePath)) {
Review Comment:
confirm this is indeed okay to remove.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/HadoopConfigUtils.java:
##########
@@ -54,36 +52,6 @@ public static Configuration createHadoopConf(Properties
props) {
*/
public static Option<String> getRawValueWithAltKeys(Configuration conf,
ConfigProperty<?>
configProperty) {
- String value = conf.get(configProperty.key());
Review Comment:
is this not called by HoodieRealtimeRecordReaderUtils ultimately? no more?
##########
hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java:
##########
@@ -175,22 +176,13 @@ public static Set<Pair<String, String>>
flattenPartitionToReplaceFileIds(Map<Str
*/
public static Option<String>
getValidCheckpointForCurrentWriter(HoodieTimeline timeline, String
checkpointKey,
String
keyToLookup) {
- return (Option<String>)
timeline.getWriteTimeline().filterCompletedInstants().getReverseOrderedInstants()
- .map(instant -> {
- try {
- HoodieCommitMetadata commitMetadata =
timeline.readCommitMetadata(instant);
- // process commits only with checkpoint entries
- String checkpointValue = commitMetadata.getMetadata(checkpointKey);
- if (StringUtils.nonEmpty(checkpointValue)) {
- // return if checkpoint for "keyForLookup" exists.
- return readCheckpointValue(checkpointValue, keyToLookup);
- } else {
- return Option.empty();
- }
- } catch (IOException e) {
- throw new HoodieIOException("Failed to parse HoodieCommitMetadata
for " + instant.toString(), e);
- }
- }).filter(Option::isPresent).findFirst().orElse(Option.empty());
+ return
TimelineUtils.findLatestInCommitMetadata(timeline.getWriteTimeline().filterCompletedInstants(),
Review Comment:
does this generate reverse ordered instants as well?
--
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]
