You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "the-other-tim-brown (via GitHub)" <gi...@apache.org> on 2025/05/27 02:46:35 UTC
[PR] [DRAFT] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
the-other-tim-brown opened a new pull request, #13361:
URL: https://github.com/apache/hudi/pull/13361
### Change Logs
- Replaces usage of HoodieFileSliceReader in Java and Spark Clustering code and in the secondary index utils with the HoodieFileGroupReader
### Impact
- Migrate old code paths to new reader for simplification and so we can consolidate performance efforts on a single reader path
### Risk level (write none, low medium or high below)
Low, these paths are covered by existing tests
### Documentation Update
_Describe any necessary documentation update if there is any new feature, config, or user-facing change. If not, put "none"._
- _The config description must be updated if new configs are added or the default value of the configs are changed_
- _Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
ticket number here and follow the [instruction](https://hudi.apache.org/contribute/developer-setup#website) to make
changes to the website._
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Change Logs and Impact were stated clearly
- [ ] Adequate tests were added if applicable
- [ ] CI passed
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2110110721
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +96,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperation2FileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperation2FileSlice(String basePath, ClusteringOperation clusteringOperation) {
Review Comment:
clusteringOperation2FileSlice: here "2" is formal?
--
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
Re: [PR] [HUDI-8284] Migrate HoodieFileSliceReader to HoodieFileGroupReader in Java and Spark engines [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2922759585
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
}, {
"hash" : "ae7da5933312996bee5726f69eb9197cc756e986",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895",
"triggerID" : "ae7da5933312996bee5726f69eb9197cc756e986",
"triggerType" : "PUSH"
}, {
"hash" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"triggerType" : "PUSH"
}, {
"hash" : "e4866d29f26c3becbf9b26c2ef4a6a31c3283f39",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e4866d29f26c3becbf9b26c2ef4a6a31c3283f39",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ae7da5933312996bee5726f69eb9197cc756e986 Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895)
* 811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a UNKNOWN
* e4866d29f26c3becbf9b26c2ef4a6a31c3283f39 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913259513
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d7a6d8e1852551f37d044a2ebc388278bd293d65 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774)
* f85013cb05ec0fbe4b991961cd6ca0456daaefea UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2114774513
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -20,36 +20,45 @@
import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.engine.ReaderContextFactory;
+import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.ClusteringOperation;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
-import org.apache.hudi.common.table.HoodieTableConfig;
-import org.apache.hudi.common.table.log.HoodieFileSliceReader;
-import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.read.HoodieFileGroupReader;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
-import org.apache.hudi.common.util.collection.CloseableMappingIterator;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieClusteringException;
-import org.apache.hudi.io.storage.HoodieFileReader;
-import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.internal.schema.InternalSchema;
+import org.apache.hudi.internal.schema.utils.SerDeHelper;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.avro.Schema;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.Serializable;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.config.HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE;
+import static org.apache.hudi.common.config.HoodieReaderConfig.MERGE_USE_RECORD_POSITIONS;
/**
* Pluggable implementation for writing data into new file groups based on ClusteringPlan.
*/
public abstract class ClusteringExecutionStrategy<T, I, K, O> implements Serializable {
Review Comment:
fair enough. not really a blocking comment. we can take it as a follow up
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java:
##########
@@ -1146,7 +1149,7 @@ public void testInflightClusteringRollbackWhenUpdatesAllowed(boolean rollbackPen
HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder(EAGER);
addConfigsForPopulateMetaFields(cfgBuilder, true);
cfgBuilder.withClusteringConfig(clusteringConfig);
- cfgBuilder.withProperties(getPropertiesForKeyGen());
+ cfgBuilder.withProperties(getPropertiesForKeyGen(true));
Review Comment:
shouldn't this be
```
cfgBuilder.withProperties(getPropertiesForKeyGen(populateMetaFields));
```
##########
hudi-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -275,25 +198,39 @@ public static HoodieData<HoodieRecord> readSecondaryKeysFromFileSlices(HoodieEng
} else {
readerSchema = tableSchema;
}
- return createSecondaryIndexGenerator(metaClient, engineType, logFilePaths, readerSchema, partition, dataFilePath, indexDefinition,
- metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""));
+ ClosableIterator<Pair<String, String>> secondaryIndexGenerator = createSecondaryIndexGenerator(readerContextFactory.getContext(), metaClient, fileSlice, readerSchema, indexDefinition,
+ metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""), props, false);
+ return new CloseableMappingIterator<>(secondaryIndexGenerator, pair -> createSecondaryIndexRecord(pair.getKey(), pair.getValue(), indexDefinition.getIndexName(), false));
});
}
- private static ClosableIterator<HoodieRecord> createSecondaryIndexGenerator(HoodieTableMetaClient metaClient,
- EngineType engineType, List<String> logFilePaths,
- Schema tableSchema, String partition,
- Option<StoragePath> dataFilePath,
- HoodieIndexDefinition indexDefinition,
- String instantTime) throws Exception {
- return new ClosableIterator<HoodieRecord>() {
- private final HoodieFileSliceReader<HoodieRecord> fileSliceReader = getFileSliceReader(
- metaClient, engineType, logFilePaths, tableSchema, partition, dataFilePath, instantTime);
- private HoodieRecord nextValidRecord;
+ private static <T> ClosableIterator<Pair<String, String>> createSecondaryIndexGenerator(HoodieReaderContext<T> readerContext,
Review Comment:
I know this method was not added in this patch.
but can we rename this to `createSecondaryIndexRecordGenerator`
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
Review Comment:
gotcha
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java:
##########
@@ -471,12 +413,7 @@ public Iterator<InternalRow> call(ClusteringOperation clusteringOperation) throw
}
// instantiate FG reader
- HoodieTableMetaClient metaClient = getHoodieTable().getMetaClient();
- HoodieReaderContext<InternalRow> readerContext = readerContextFactory.getContext();
- HoodieFileGroupReader<InternalRow> fileGroupReader = HoodieFileGroupReader.<InternalRow>newBuilder()
- .withReaderContext(readerContext).withHoodieTableMetaClient(metaClient).withLatestCommitTime(instantTime)
- .withFileSlice(fileSlice).withDataSchema(readerSchema).withRequestedSchema(readerSchema).withInternalSchema(internalSchemaOption)
- .withShouldUseRecordPosition(usePosition).withProps(metaClient.getTableConfig().getProps()).build();
+ HoodieFileGroupReader<InternalRow> fileGroupReader = getFileGroupReader(metaClient, fileSlice, readerSchema, internalSchemaOption, readerContextFactory, instantTime, usePosition);
Review Comment:
cool
##########
hudi-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -275,25 +198,39 @@ public static HoodieData<HoodieRecord> readSecondaryKeysFromFileSlices(HoodieEng
} else {
readerSchema = tableSchema;
}
- return createSecondaryIndexGenerator(metaClient, engineType, logFilePaths, readerSchema, partition, dataFilePath, indexDefinition,
- metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""));
+ ClosableIterator<Pair<String, String>> secondaryIndexGenerator = createSecondaryIndexGenerator(readerContextFactory.getContext(), metaClient, fileSlice, readerSchema, indexDefinition,
+ metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""), props, false);
+ return new CloseableMappingIterator<>(secondaryIndexGenerator, pair -> createSecondaryIndexRecord(pair.getKey(), pair.getValue(), indexDefinition.getIndexName(), false));
});
}
- private static ClosableIterator<HoodieRecord> createSecondaryIndexGenerator(HoodieTableMetaClient metaClient,
- EngineType engineType, List<String> logFilePaths,
- Schema tableSchema, String partition,
- Option<StoragePath> dataFilePath,
- HoodieIndexDefinition indexDefinition,
- String instantTime) throws Exception {
- return new ClosableIterator<HoodieRecord>() {
- private final HoodieFileSliceReader<HoodieRecord> fileSliceReader = getFileSliceReader(
- metaClient, engineType, logFilePaths, tableSchema, partition, dataFilePath, instantTime);
- private HoodieRecord nextValidRecord;
+ private static <T> ClosableIterator<Pair<String, String>> createSecondaryIndexGenerator(HoodieReaderContext<T> readerContext,
Review Comment:
java docs. what do we return from here.
each constituent of the Pair specifically
--
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
Re: [PR] [HUDI-8284] Migrate HoodieFileSliceReader to HoodieFileGroupReader in Java and Spark engines [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2923053853
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
}, {
"hash" : "ae7da5933312996bee5726f69eb9197cc756e986",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895",
"triggerID" : "ae7da5933312996bee5726f69eb9197cc756e986",
"triggerType" : "PUSH"
}, {
"hash" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"triggerType" : "PUSH"
}, {
"hash" : "e4866d29f26c3becbf9b26c2ef4a6a31c3283f39",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5897",
"triggerID" : "e4866d29f26c3becbf9b26c2ef4a6a31c3283f39",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a UNKNOWN
* e4866d29f26c3becbf9b26c2ef4a6a31c3283f39 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5897)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2114824081
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
+ boolean baseFileExists = !StringUtils.isNullOrEmpty(clusteringOperation.getDataFilePath());
+ HoodieBaseFile baseFile = baseFileExists ? new HoodieBaseFile(new StoragePath(basePath, clusteringOperation.getDataFilePath()).toString()) : null;
+ List<HoodieLogFile> logFiles = clusteringOperation.getDeltaFilePaths().stream().map(p ->
+ new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
+ basePath, partitionPath), p)))
+ .sorted(new HoodieLogFile.LogFileComparator())
+ .collect(Collectors.toList());
+
+ ValidationUtils.checkState(baseFileExists || !logFiles.isEmpty(), "Both base file and log files are missing from this clustering operation " + clusteringOperation);
+ String baseInstantTime = baseFileExists ? baseFile.getCommitTime() : logFiles.get(0).getDeltaCommitTime();
+ FileSlice fileSlice = new FileSlice(partitionPath, baseInstantTime, clusteringOperation.getFileId());
+ fileSlice.setBaseFile(baseFile);
Review Comment:
Good point, updated this to use option and also handle the bootstrap file if present
##########
hudi-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -275,25 +198,39 @@ public static HoodieData<HoodieRecord> readSecondaryKeysFromFileSlices(HoodieEng
} else {
readerSchema = tableSchema;
}
- return createSecondaryIndexGenerator(metaClient, engineType, logFilePaths, readerSchema, partition, dataFilePath, indexDefinition,
- metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""));
+ ClosableIterator<Pair<String, String>> secondaryIndexGenerator = createSecondaryIndexGenerator(readerContextFactory.getContext(), metaClient, fileSlice, readerSchema, indexDefinition,
+ metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""), props, false);
+ return new CloseableMappingIterator<>(secondaryIndexGenerator, pair -> createSecondaryIndexRecord(pair.getKey(), pair.getValue(), indexDefinition.getIndexName(), false));
});
}
- private static ClosableIterator<HoodieRecord> createSecondaryIndexGenerator(HoodieTableMetaClient metaClient,
- EngineType engineType, List<String> logFilePaths,
- Schema tableSchema, String partition,
- Option<StoragePath> dataFilePath,
- HoodieIndexDefinition indexDefinition,
- String instantTime) throws Exception {
- return new ClosableIterator<HoodieRecord>() {
- private final HoodieFileSliceReader<HoodieRecord> fileSliceReader = getFileSliceReader(
- metaClient, engineType, logFilePaths, tableSchema, partition, dataFilePath, instantTime);
- private HoodieRecord nextValidRecord;
+ private static <T> ClosableIterator<Pair<String, String>> createSecondaryIndexGenerator(HoodieReaderContext<T> readerContext,
Review Comment:
Yes, updated
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913838875
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb63815321beecb1d65ad17bc2f16961e9b159d7 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2110124866
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +96,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperation2FileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperation2FileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
+ boolean baseFileExists = !StringUtils.isNullOrEmpty(clusteringOperation.getDataFilePath());
+ HoodieBaseFile baseFile = baseFileExists ? new HoodieBaseFile(new StoragePath(basePath, clusteringOperation.getDataFilePath()).toString()) : null;
+ List<HoodieLogFile> logFiles = clusteringOperation.getDeltaFilePaths().stream().map(p ->
+ new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
+ basePath, partitionPath), p)))
+ .sorted(new HoodieLogFile.LogFileComparator())
+ .collect(Collectors.toList());
+
+ ValidationUtils.checkState(baseFileExists || !logFiles.isEmpty(), "Both base file and log files are missing from this clustering operation " + clusteringOperation);
+ String baseInstantTime = baseFileExists ? baseFile.getCommitTime() : logFiles.get(0).getDeltaCommitTime();
+ FileSlice fileSlice = new FileSlice(partitionPath, baseInstantTime, clusteringOperation.getFileId());
+ fileSlice.setBaseFile(baseFile);
+ logFiles.forEach(fileSlice::addLogFile);
+ return fileSlice;
+ }
+
+ protected static <R> HoodieFileGroupReader<R> getFileGroupReader(HoodieTableMetaClient metaClient, FileSlice fileSlice, Schema readerSchema, Option<InternalSchema> internalSchemaOption,
Review Comment:
Here R and T are the same or not?
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2920710368
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 316c2fc25ddbc5349f5e604214f3ede309bf48e9 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786)
* 3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863)
* c484cb18b09579fb36a643f9d00b1e3c8309ddbe UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] Migrate HoodieFileSliceReader to HoodieFileGroupReader in Java and Spark engines [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2922738823
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
}, {
"hash" : "ae7da5933312996bee5726f69eb9197cc756e986",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895",
"triggerID" : "ae7da5933312996bee5726f69eb9197cc756e986",
"triggerType" : "PUSH"
}, {
"hash" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c484cb18b09579fb36a643f9d00b1e3c8309ddbe Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865)
* ae7da5933312996bee5726f69eb9197cc756e986 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895)
* 811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913474639
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 682d287b40cf2e8fbe05e6605934e53c3ce1863e Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781)
* eb63815321beecb1d65ad17bc2f16961e9b159d7 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2110190093
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +96,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperation2FileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperation2FileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
+ boolean baseFileExists = !StringUtils.isNullOrEmpty(clusteringOperation.getDataFilePath());
+ HoodieBaseFile baseFile = baseFileExists ? new HoodieBaseFile(new StoragePath(basePath, clusteringOperation.getDataFilePath()).toString()) : null;
+ List<HoodieLogFile> logFiles = clusteringOperation.getDeltaFilePaths().stream().map(p ->
+ new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
+ basePath, partitionPath), p)))
+ .sorted(new HoodieLogFile.LogFileComparator())
+ .collect(Collectors.toList());
+
+ ValidationUtils.checkState(baseFileExists || !logFiles.isEmpty(), "Both base file and log files are missing from this clustering operation " + clusteringOperation);
+ String baseInstantTime = baseFileExists ? baseFile.getCommitTime() : logFiles.get(0).getDeltaCommitTime();
+ FileSlice fileSlice = new FileSlice(partitionPath, baseInstantTime, clusteringOperation.getFileId());
+ fileSlice.setBaseFile(baseFile);
+ logFiles.forEach(fileSlice::addLogFile);
+ return fileSlice;
+ }
+
+ protected static <R> HoodieFileGroupReader<R> getFileGroupReader(HoodieTableMetaClient metaClient, FileSlice fileSlice, Schema readerSchema, Option<InternalSchema> internalSchemaOption,
Review Comment:
It is a static method so it is not tied to the instance's `T` so I wanted to use a different generic to ensure that was clear.
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2920702014
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 316c2fc25ddbc5349f5e604214f3ede309bf48e9 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786)
* 3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2920699069
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 316c2fc25ddbc5349f5e604214f3ede309bf48e9 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786)
* 3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2115072308
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
+ boolean baseFileExists = !StringUtils.isNullOrEmpty(clusteringOperation.getDataFilePath());
+ HoodieBaseFile baseFile = baseFileExists ? new HoodieBaseFile(new StoragePath(basePath, clusteringOperation.getDataFilePath()).toString()) : null;
+ List<HoodieLogFile> logFiles = clusteringOperation.getDeltaFilePaths().stream().map(p ->
+ new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
+ basePath, partitionPath), p)))
+ .sorted(new HoodieLogFile.LogFileComparator())
+ .collect(Collectors.toList());
+
+ ValidationUtils.checkState(baseFileExists || !logFiles.isEmpty(), "Both base file and log files are missing from this clustering operation " + clusteringOperation);
+ String baseInstantTime = baseFileExists ? baseFile.getCommitTime() : logFiles.get(0).getDeltaCommitTime();
+ FileSlice fileSlice = new FileSlice(partitionPath, baseInstantTime, clusteringOperation.getFileId());
+ fileSlice.setBaseFile(baseFile);
Review Comment:
hmmm, this is what I was actually looking for and did not see anything.
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2115072463
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
+ boolean baseFileExists = !StringUtils.isNullOrEmpty(clusteringOperation.getDataFilePath());
+ HoodieBaseFile baseFile = baseFileExists ? new HoodieBaseFile(new StoragePath(basePath, clusteringOperation.getDataFilePath()).toString()) : null;
+ List<HoodieLogFile> logFiles = clusteringOperation.getDeltaFilePaths().stream().map(p ->
+ new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
+ basePath, partitionPath), p)))
+ .sorted(new HoodieLogFile.LogFileComparator())
+ .collect(Collectors.toList());
+
+ ValidationUtils.checkState(baseFileExists || !logFiles.isEmpty(), "Both base file and log files are missing from this clustering operation " + clusteringOperation);
+ String baseInstantTime = baseFileExists ? baseFile.getCommitTime() : logFiles.get(0).getDeltaCommitTime();
+ FileSlice fileSlice = new FileSlice(partitionPath, baseInstantTime, clusteringOperation.getFileId());
+ fileSlice.setBaseFile(baseFile);
Review Comment:
do we have tests for this. how come we did not catch it
--
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
Re: [PR] [DRAFT] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2911030378
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 430091cd9076397f72710bd0fcf2616b23c855d2 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2920784780
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863)
* c484cb18b09579fb36a643f9d00b1e3c8309ddbe Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] Migrate HoodieFileSliceReader to HoodieFileGroupReader in Java and Spark engines [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2922713650
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
}, {
"hash" : "ae7da5933312996bee5726f69eb9197cc756e986",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ae7da5933312996bee5726f69eb9197cc756e986",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c484cb18b09579fb36a643f9d00b1e3c8309ddbe Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865)
* ae7da5933312996bee5726f69eb9197cc756e986 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913420544
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f85013cb05ec0fbe4b991961cd6ca0456daaefea Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779)
* 682d287b40cf2e8fbe05e6605934e53c3ce1863e Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913400861
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f85013cb05ec0fbe4b991961cd6ca0456daaefea Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779)
* 682d287b40cf2e8fbe05e6605934e53c3ce1863e UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [DRAFT] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2912614011
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 430091cd9076397f72710bd0fcf2616b23c855d2 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755)
* d7a6d8e1852551f37d044a2ebc388278bd293d65 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2110132858
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java:
##########
@@ -456,7 +397,8 @@ private Dataset<Row> readRecordsForGroupAsRowWithFileGroupReader(JavaSparkContex
SerializableSchema serializableTableSchemaWithMetaFields = new SerializableSchema(tableSchemaWithMetaFields);
// broadcast reader context.
- ReaderContextFactory<InternalRow> readerContextFactory = getEngineContext().getReaderContextFactory(getHoodieTable().getMetaClient());
+ HoodieTableMetaClient metaClient = getHoodieTable().getMetaClient();
+ ReaderContextFactory<InternalRow> readerContextFactory = getEngineContext().getReaderContextFactory(metaClient);
Review Comment:
When do we use T, when use InternalRow?
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913396365
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d7a6d8e1852551f37d044a2ebc388278bd293d65 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774)
* f85013cb05ec0fbe4b991961cd6ca0456daaefea Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779)
* 682d287b40cf2e8fbe05e6605934e53c3ce1863e UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913469806
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f85013cb05ec0fbe4b991961cd6ca0456daaefea Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779)
* 682d287b40cf2e8fbe05e6605934e53c3ce1863e Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781)
* eb63815321beecb1d65ad17bc2f16961e9b159d7 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2114260663
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java:
##########
@@ -471,12 +413,7 @@ public Iterator<InternalRow> call(ClusteringOperation clusteringOperation) throw
}
// instantiate FG reader
- HoodieTableMetaClient metaClient = getHoodieTable().getMetaClient();
- HoodieReaderContext<InternalRow> readerContext = readerContextFactory.getContext();
- HoodieFileGroupReader<InternalRow> fileGroupReader = HoodieFileGroupReader.<InternalRow>newBuilder()
- .withReaderContext(readerContext).withHoodieTableMetaClient(metaClient).withLatestCommitTime(instantTime)
- .withFileSlice(fileSlice).withDataSchema(readerSchema).withRequestedSchema(readerSchema).withInternalSchema(internalSchemaOption)
- .withShouldUseRecordPosition(usePosition).withProps(metaClient.getTableConfig().getProps()).build();
+ HoodieFileGroupReader<InternalRow> fileGroupReader = getFileGroupReader(metaClient, fileSlice, readerSchema, internalSchemaOption, readerContextFactory, instantTime, usePosition);
Review Comment:
may be I am not aware of this.
does this seamlessly take care of bootstrap base path as well?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
Review Comment:
I feel constructing the FileSlice based on a given clusteringOperation might be repeated thing. given HoodieFileGroupReader takes in a FileSlice as an argument.
can we move this to some common utility method. depending on bandwidth, we can punt fixing all other callers (outside of this patch) for now. but atleast file a tracking ticket and refer to the common utility method that we might add it as part of this patch.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
+ boolean baseFileExists = !StringUtils.isNullOrEmpty(clusteringOperation.getDataFilePath());
+ HoodieBaseFile baseFile = baseFileExists ? new HoodieBaseFile(new StoragePath(basePath, clusteringOperation.getDataFilePath()).toString()) : null;
+ List<HoodieLogFile> logFiles = clusteringOperation.getDeltaFilePaths().stream().map(p ->
+ new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
+ basePath, partitionPath), p)))
+ .sorted(new HoodieLogFile.LogFileComparator())
+ .collect(Collectors.toList());
+
+ ValidationUtils.checkState(baseFileExists || !logFiles.isEmpty(), "Both base file and log files are missing from this clustering operation " + clusteringOperation);
+ String baseInstantTime = baseFileExists ? baseFile.getCommitTime() : logFiles.get(0).getDeltaCommitTime();
+ FileSlice fileSlice = new FileSlice(partitionPath, baseInstantTime, clusteringOperation.getFileId());
+ fileSlice.setBaseFile(baseFile);
Review Comment:
baseFile could be null as well.
can we maintain Option earlier and do option.ifPresent here
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
+ boolean baseFileExists = !StringUtils.isNullOrEmpty(clusteringOperation.getDataFilePath());
+ HoodieBaseFile baseFile = baseFileExists ? new HoodieBaseFile(new StoragePath(basePath, clusteringOperation.getDataFilePath()).toString()) : null;
+ List<HoodieLogFile> logFiles = clusteringOperation.getDeltaFilePaths().stream().map(p ->
+ new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
+ basePath, partitionPath), p)))
+ .sorted(new HoodieLogFile.LogFileComparator())
Review Comment:
https://github.com/apache/hudi/blob/94e4d3763e7919c674762625ea8f331629691ad7/hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java#L125
we already enforce ordering within FG reader.
can we avoid the duplicate ordering
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
Review Comment:
or if we taking care of all such callers in this patch, I am good.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -20,36 +20,45 @@
import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.engine.ReaderContextFactory;
+import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.ClusteringOperation;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
-import org.apache.hudi.common.table.HoodieTableConfig;
-import org.apache.hudi.common.table.log.HoodieFileSliceReader;
-import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.read.HoodieFileGroupReader;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
-import org.apache.hudi.common.util.collection.CloseableMappingIterator;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieClusteringException;
-import org.apache.hudi.io.storage.HoodieFileReader;
-import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.internal.schema.InternalSchema;
+import org.apache.hudi.internal.schema.utils.SerDeHelper;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.avro.Schema;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.Serializable;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.config.HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE;
+import static org.apache.hudi.common.config.HoodieReaderConfig.MERGE_USE_RECORD_POSITIONS;
/**
* Pluggable implementation for writing data into new file groups based on ClusteringPlan.
*/
public abstract class ClusteringExecutionStrategy<T, I, K, O> implements Serializable {
Review Comment:
should we rename this class to `BaseClusteringExecutionStrategy`
--
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
Re: [PR] [HUDI-8284] Migrate HoodieFileSliceReader to HoodieFileGroupReader in Java and Spark engines [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2922717850
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
}, {
"hash" : "ae7da5933312996bee5726f69eb9197cc756e986",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895",
"triggerID" : "ae7da5933312996bee5726f69eb9197cc756e986",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c484cb18b09579fb36a643f9d00b1e3c8309ddbe Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865)
* ae7da5933312996bee5726f69eb9197cc756e986 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [DRAFT] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2910954546
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 430091cd9076397f72710bd0fcf2616b23c855d2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2110191547
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java:
##########
@@ -456,7 +397,8 @@ private Dataset<Row> readRecordsForGroupAsRowWithFileGroupReader(JavaSparkContex
SerializableSchema serializableTableSchemaWithMetaFields = new SerializableSchema(tableSchemaWithMetaFields);
// broadcast reader context.
- ReaderContextFactory<InternalRow> readerContextFactory = getEngineContext().getReaderContextFactory(getHoodieTable().getMetaClient());
+ HoodieTableMetaClient metaClient = getHoodieTable().getMetaClient();
+ ReaderContextFactory<InternalRow> readerContextFactory = getEngineContext().getReaderContextFactory(metaClient);
Review Comment:
T is a generic we use when we do not know the specific type but in this case since it is a spark specific implementation, we know that it will be `InternalRow`
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913579290
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 682d287b40cf2e8fbe05e6605934e53c3ce1863e Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781)
* eb63815321beecb1d65ad17bc2f16961e9b159d7 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [DRAFT] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2109301311
##########
hudi-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -275,25 +197,39 @@ public static HoodieData<HoodieRecord> readSecondaryKeysFromFileSlices(HoodieEng
} else {
readerSchema = tableSchema;
}
- return createSecondaryIndexGenerator(metaClient, engineType, logFilePaths, readerSchema, partition, dataFilePath, indexDefinition,
- metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""));
+ ClosableIterator<Pair<String, String>> secondaryIndexGenerator = createSecondaryIndexGenerator(readerContextFactory.getContext(), metaClient, fileSlice, readerSchema, indexDefinition,
+ metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""), props, false);
+ return new CloseableMappingIterator<>(secondaryIndexGenerator, pair -> createSecondaryIndexRecord(pair.getKey(), pair.getValue(), partition, false));
});
}
- private static ClosableIterator<HoodieRecord> createSecondaryIndexGenerator(HoodieTableMetaClient metaClient,
- EngineType engineType, List<String> logFilePaths,
- Schema tableSchema, String partition,
- Option<StoragePath> dataFilePath,
- HoodieIndexDefinition indexDefinition,
- String instantTime) throws Exception {
- return new ClosableIterator<HoodieRecord>() {
- private final HoodieFileSliceReader<HoodieRecord> fileSliceReader = getFileSliceReader(
- metaClient, engineType, logFilePaths, tableSchema, partition, dataFilePath, instantTime);
- private HoodieRecord nextValidRecord;
+ private static <T> ClosableIterator<Pair<String, String>> createSecondaryIndexGenerator(HoodieReaderContext<T> readerContext,
+ HoodieTableMetaClient metaClient,
+ FileSlice fileSlice,
+ Schema tableSchema,
+ HoodieIndexDefinition indexDefinition,
+ String instantTime,
+ TypedProperties props,
+ boolean allowInflightInstants) throws Exception {
+ String secondaryKeyField = String.join(".", indexDefinition.getSourceFields());
+ HoodieFileGroupReader<T> fileGroupReader = HoodieFileGroupReader.<T>newBuilder()
+ .withReaderContext(readerContext)
+ .withFileSlice(fileSlice)
+ .withHoodieTableMetaClient(metaClient)
+ .withProps(props)
+ .withLatestCommitTime(instantTime)
+ .withDataSchema(tableSchema)
+ .withRequestedSchema(tableSchema)
Review Comment:
We can just read the required fields here but we don't have a clean utility to construct the optimal schema yet so I have added the context to this existing ticket: https://issues.apache.org/jira/browse/HUDI-9439
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2914316333
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 316c2fc25ddbc5349f5e604214f3ede309bf48e9 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2114824284
##########
hudi-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -275,25 +198,39 @@ public static HoodieData<HoodieRecord> readSecondaryKeysFromFileSlices(HoodieEng
} else {
readerSchema = tableSchema;
}
- return createSecondaryIndexGenerator(metaClient, engineType, logFilePaths, readerSchema, partition, dataFilePath, indexDefinition,
- metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""));
+ ClosableIterator<Pair<String, String>> secondaryIndexGenerator = createSecondaryIndexGenerator(readerContextFactory.getContext(), metaClient, fileSlice, readerSchema, indexDefinition,
+ metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(""), props, false);
+ return new CloseableMappingIterator<>(secondaryIndexGenerator, pair -> createSecondaryIndexRecord(pair.getKey(), pair.getValue(), indexDefinition.getIndexName(), false));
});
}
- private static ClosableIterator<HoodieRecord> createSecondaryIndexGenerator(HoodieTableMetaClient metaClient,
- EngineType engineType, List<String> logFilePaths,
- Schema tableSchema, String partition,
- Option<StoragePath> dataFilePath,
- HoodieIndexDefinition indexDefinition,
- String instantTime) throws Exception {
- return new ClosableIterator<HoodieRecord>() {
- private final HoodieFileSliceReader<HoodieRecord> fileSliceReader = getFileSliceReader(
- metaClient, engineType, logFilePaths, tableSchema, partition, dataFilePath, instantTime);
- private HoodieRecord nextValidRecord;
+ private static <T> ClosableIterator<Pair<String, String>> createSecondaryIndexGenerator(HoodieReaderContext<T> readerContext,
Review Comment:
Added doc
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2920713213
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863)
* c484cb18b09579fb36a643f9d00b1e3c8309ddbe UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [DRAFT] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2910952757
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 430091cd9076397f72710bd0fcf2616b23c855d2 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913264173
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d7a6d8e1852551f37d044a2ebc388278bd293d65 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774)
* f85013cb05ec0fbe4b991961cd6ca0456daaefea Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2914058386
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb63815321beecb1d65ad17bc2f16961e9b159d7 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782)
* 316c2fc25ddbc5349f5e604214f3ede309bf48e9 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] Migrate HoodieFileSliceReader to HoodieFileGroupReader in Java and Spark engines [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2922742877
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
}, {
"hash" : "ae7da5933312996bee5726f69eb9197cc756e986",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895",
"triggerID" : "ae7da5933312996bee5726f69eb9197cc756e986",
"triggerType" : "PUSH"
}, {
"hash" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ae7da5933312996bee5726f69eb9197cc756e986 Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895)
* 811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] Migrate HoodieFileSliceReader to HoodieFileGroupReader in Java and Spark engines [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2922849827
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
}, {
"hash" : "ae7da5933312996bee5726f69eb9197cc756e986",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895",
"triggerID" : "ae7da5933312996bee5726f69eb9197cc756e986",
"triggerType" : "PUSH"
}, {
"hash" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a",
"triggerType" : "PUSH"
}, {
"hash" : "e4866d29f26c3becbf9b26c2ef4a6a31c3283f39",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5897",
"triggerID" : "e4866d29f26c3becbf9b26c2ef4a6a31c3283f39",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ae7da5933312996bee5726f69eb9197cc756e986 Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5895)
* 811f5a0b0ad82a91d9b86e5cb859fc99f1d9227a UNKNOWN
* e4866d29f26c3becbf9b26c2ef4a6a31c3283f39 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5897)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2114280118
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
Review Comment:
This method is actually just copied from another class so it can be used as a common utility here. It is not new code: https://github.com/apache/hudi/pull/13361/files#diff-4ec865bde17bc071ff44da1d50995153e5887acb4b2ff379f1f644df26b85994L492
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2114281883
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java:
##########
@@ -471,12 +413,7 @@ public Iterator<InternalRow> call(ClusteringOperation clusteringOperation) throw
}
// instantiate FG reader
- HoodieTableMetaClient metaClient = getHoodieTable().getMetaClient();
- HoodieReaderContext<InternalRow> readerContext = readerContextFactory.getContext();
- HoodieFileGroupReader<InternalRow> fileGroupReader = HoodieFileGroupReader.<InternalRow>newBuilder()
- .withReaderContext(readerContext).withHoodieTableMetaClient(metaClient).withLatestCommitTime(instantTime)
- .withFileSlice(fileSlice).withDataSchema(readerSchema).withRequestedSchema(readerSchema).withInternalSchema(internalSchemaOption)
- .withShouldUseRecordPosition(usePosition).withProps(metaClient.getTableConfig().getProps()).build();
+ HoodieFileGroupReader<InternalRow> fileGroupReader = getFileGroupReader(metaClient, fileSlice, readerSchema, internalSchemaOption, readerContextFactory, instantTime, usePosition);
Review Comment:
Yes, the FGReader now handles bootstrap 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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2110111532
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +96,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperation2FileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperation2FileSlice(String basePath, ClusteringOperation clusteringOperation) {
Review Comment:
I copied this from the other class but I will change it since I am not a fan of it :)
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2114281337
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -20,36 +20,45 @@
import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.engine.ReaderContextFactory;
+import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.ClusteringOperation;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
-import org.apache.hudi.common.table.HoodieTableConfig;
-import org.apache.hudi.common.table.log.HoodieFileSliceReader;
-import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.read.HoodieFileGroupReader;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
-import org.apache.hudi.common.util.collection.CloseableMappingIterator;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieClusteringException;
-import org.apache.hudi.io.storage.HoodieFileReader;
-import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.internal.schema.InternalSchema;
+import org.apache.hudi.internal.schema.utils.SerDeHelper;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.avro.Schema;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.Serializable;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.config.HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE;
+import static org.apache.hudi.common.config.HoodieReaderConfig.MERGE_USE_RECORD_POSITIONS;
/**
* Pluggable implementation for writing data into new file groups based on ClusteringPlan.
*/
public abstract class ClusteringExecutionStrategy<T, I, K, O> implements Serializable {
Review Comment:
I don't know if it is worth renaming the classes as part of this
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2914053550
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb63815321beecb1d65ad17bc2f16961e9b159d7 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782)
* 316c2fc25ddbc5349f5e604214f3ede309bf48e9 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2114824998
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java:
##########
@@ -1146,7 +1149,7 @@ public void testInflightClusteringRollbackWhenUpdatesAllowed(boolean rollbackPen
HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder(EAGER);
addConfigsForPopulateMetaFields(cfgBuilder, true);
cfgBuilder.withClusteringConfig(clusteringConfig);
- cfgBuilder.withProperties(getPropertiesForKeyGen());
+ cfgBuilder.withProperties(getPropertiesForKeyGen(true));
Review Comment:
this case does not have `populateMetaFields` defined
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2913010082
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d7a6d8e1852551f37d044a2ebc388278bd293d65 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8284] Migrate HoodieFileSliceReader to HoodieFileGroupReader in Java and Spark engines [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan merged PR #13361:
URL: https://github.com/apache/hudi/pull/13361
--
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
Re: [PR] [HUDI-8284] Migrate HoodieFileSliceReader to HoodieFileGroupReader in Java and Spark engines [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13361:
URL: https://github.com/apache/hudi/pull/13361#discussion_r2116138598
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java:
##########
@@ -84,51 +93,50 @@ protected HoodieWriteConfig getWriteConfig() {
return this.writeConfig;
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithLogFiles(ClusteringOperation operation, String instantTime, long maxMemory,
- Option<BaseKeyGenerator> keyGeneratorOpt, Option<HoodieFileReader> baseFileReaderOpt) {
+ protected ClosableIterator<HoodieRecord<T>> getRecordIterator(ReaderContextFactory<T> readerContextFactory, ClusteringOperation operation, String instantTime, long maxMemory) {
HoodieWriteConfig config = getWriteConfig();
+ TypedProperties props = TypedProperties.copy(config.getProps());
+ props.setProperty(MAX_MEMORY_FOR_MERGE.key(), Long.toString(maxMemory));
+
HoodieTable table = getHoodieTable();
- HoodieTableConfig tableConfig = table.getMetaClient().getTableConfig();
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(table.getStorage())
- .withBasePath(table.getMetaClient().getBasePath())
- .withLogFilePaths(operation.getDeltaFilePaths())
- .withReaderSchema(readerSchemaWithMetaFields)
- .withLatestInstantTime(instantTime)
- .withMaxMemorySizeInBytes(maxMemory)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(config.enableOptimizedLogBlocksScan())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(table.getMetaClient())
- .build();
+ FileSlice fileSlice = clusteringOperationToFileSlice(table.getMetaClient().getBasePath().toString(), operation);
+ final boolean usePosition = getWriteConfig().getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchema = SerDeHelper.fromJson(getWriteConfig().getInternalSchema());
try {
- return new HoodieFileSliceReader(baseFileReaderOpt, scanner, readerSchemaWithMetaFields, tableConfig.getPreCombineField(), config.getRecordMerger(),
- tableConfig.getProps(),
- tableConfig.populateMetaFields() ? Option.empty() : Option.of(Pair.of(tableConfig.getRecordKeyFieldProp(),
- tableConfig.getPartitionFieldProp())), keyGeneratorOpt);
+ return getFileGroupReader(table.getMetaClient(), fileSlice, readerSchemaWithMetaFields, internalSchema, readerContextFactory, instantTime, usePosition).getClosableHoodieRecordIterator();
} catch (IOException e) {
throw new HoodieClusteringException("Error reading file slices", e);
}
}
- protected ClosableIterator<HoodieRecord<T>> getRecordIteratorWithBaseFileOnly(Option<BaseKeyGenerator> keyGeneratorOpt, HoodieFileReader baseFileReader) {
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be shuffled later.
- ClosableIterator<HoodieRecord> baseRecordsIterator;
- try {
- baseRecordsIterator = baseFileReader.getRecordIterator(readerSchemaWithMetaFields);
- } catch (IOException e) {
- throw new HoodieClusteringException("Error reading base file", e);
- }
- return new CloseableMappingIterator(
- baseRecordsIterator,
- rec -> ((HoodieRecord) rec).copy().wrapIntoHoodieRecordPayloadWithKeyGen(readerSchemaWithMetaFields, writeConfig.getProps(), keyGeneratorOpt));
+ /**
+ * Construct FileSlice from a given clustering operation {@code clusteringOperation}.
+ */
+ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOperation clusteringOperation) {
+ String partitionPath = clusteringOperation.getPartitionPath();
+ boolean baseFileExists = !StringUtils.isNullOrEmpty(clusteringOperation.getDataFilePath());
+ HoodieBaseFile baseFile = baseFileExists ? new HoodieBaseFile(new StoragePath(basePath, clusteringOperation.getDataFilePath()).toString()) : null;
+ List<HoodieLogFile> logFiles = clusteringOperation.getDeltaFilePaths().stream().map(p ->
+ new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
+ basePath, partitionPath), p)))
+ .sorted(new HoodieLogFile.LogFileComparator())
+ .collect(Collectors.toList());
+
+ ValidationUtils.checkState(baseFileExists || !logFiles.isEmpty(), "Both base file and log files are missing from this clustering operation " + clusteringOperation);
+ String baseInstantTime = baseFileExists ? baseFile.getCommitTime() : logFiles.get(0).getDeltaCommitTime();
+ FileSlice fileSlice = new FileSlice(partitionPath, baseInstantTime, clusteringOperation.getFileId());
+ fileSlice.setBaseFile(baseFile);
Review Comment:
Previously the FileGroupReader path was gated by a check for bootstrap base files. Now I have removed that check since it is no longer required.
There is at least one existing test that will cover this [here](https://github.com/apache/hudi/blob/master/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestDataSourceForBootstrap.scala#L338)
--
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
Re: [PR] [HUDI-8284] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2920984264
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5774",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
}, {
"hash" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5779",
"triggerID" : "f85013cb05ec0fbe4b991961cd6ca0456daaefea",
"triggerType" : "PUSH"
}, {
"hash" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5781",
"triggerID" : "682d287b40cf2e8fbe05e6605934e53c3ce1863e",
"triggerType" : "PUSH"
}, {
"hash" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5782",
"triggerID" : "eb63815321beecb1d65ad17bc2f16961e9b159d7",
"triggerType" : "PUSH"
}, {
"hash" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5786",
"triggerID" : "316c2fc25ddbc5349f5e604214f3ede309bf48e9",
"triggerType" : "PUSH"
}, {
"hash" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5863",
"triggerID" : "3302d8aab65aa6da8003c3f5ea45f1a5f78e68eb",
"triggerType" : "PUSH"
}, {
"hash" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865",
"triggerID" : "c484cb18b09579fb36a643f9d00b1e3c8309ddbe",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c484cb18b09579fb36a643f9d00b1e3c8309ddbe Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5865)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [DRAFT] remove usage of HoodieFileSliceReader in java/spark paths [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13361:
URL: https://github.com/apache/hudi/pull/13361#issuecomment-2912606052
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755",
"triggerID" : "430091cd9076397f72710bd0fcf2616b23c855d2",
"triggerType" : "PUSH"
}, {
"hash" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d7a6d8e1852551f37d044a2ebc388278bd293d65",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 430091cd9076397f72710bd0fcf2616b23c855d2 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=5755)
* d7a6d8e1852551f37d044a2ebc388278bd293d65 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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