cshuo commented on code in PR #19118:
URL: https://github.com/apache/hudi/pull/19118#discussion_r3526199482
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeCDCFileWriter.java:
##########
@@ -120,14 +131,16 @@ void close() throws IOException {
}
}
+ @SuppressWarnings("unchecked")
private void ensureCDCWriter() throws IOException {
if (cdcWriter != null && cdcWriter.canWrite()) {
return;
}
close();
HoodieLogFile cdcLogFile = createNativeCDCLogFile();
- cdcWriter = HoodieFileWriterFactory.getFileWriter(
+ cdcWriter = (HoodieFileWriter<T>) HoodieFileWriterFactory.getFileWriter(
commitTime, cdcLogFile.getPath(), storage, config, cdcSchema,
taskContextSupplier, recordType);
+
cdcWriter.addFooterMetadata(NativeLogFooterMetadata.toFooterMetadata(cdcDataBlockHeader));
Review Comment:
rename `cdcDataBlockHeader`
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/CommonClientUtils.java:
##########
@@ -120,6 +131,41 @@ public static HoodieLogBlock.HoodieLogBlockType
getLogBlockType(HoodieWriteConfi
}
}
+ /**
+ * Whether log blocks should be written in the native (v2) log format
(standalone native
+ * files written via {@code HoodieNativeLogFormatWriter}) instead of the
legacy inline
+ * log format. The native format is the default for write version >=
{@link HoodieTableVersion#TEN},
+ * except for Lance base files.
+ *
+ * <p>This decision is keyed on the effective write version (i.e. {@code
HoodieWriteConfig#getWriteVersion()})
+ * and the effective base file format, consistent with how the inline log
block layout is
+ * selected, so that the on-disk format follows what the writer is targeting
during
+ * upgrade/downgrade windows. Lance remains on the legacy inline log format
until native Lance log
+ * support is complete.
+ *
+ * @param writeConfig the writer configuration.
+ * @param tableConfig the persisted table configuration.
+ */
+ public static boolean shouldWriteNativeLogs(HoodieWriteConfig writeConfig,
HoodieTableConfig tableConfig) {
Review Comment:
todo: issue, remove table config.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieCDCLogWriterFactory.java:
##########
@@ -79,7 +80,7 @@ public static <T, I, K, O> HoodieCDCLogWriter<IndexedRecord>
createAvroCDCLogWri
IOUtils.getMaxMemoryPerPartitionMerge(taskContextSupplier, config));
}
- public static boolean shouldWriteNativeCDCLogs(HoodieTableConfig
tableConfig) {
- return tableConfig.isLSMTreeStorageLayout();
+ public static boolean shouldWriteNativeCDCLogs(HoodieWriteConfig
writeConfig, HoodieTableConfig tableConfig) {
Review Comment:
remove this method.
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/WriteStatMerger.java:
##########
@@ -125,18 +125,18 @@ private static HoodieWriteStat.RuntimeStats
getMergedRuntimeStats(
return runtimeStats;
}
- private static Map<String, Long> getMergedCdcStats(Map<String, Long>
cdcStats1, Map<String, Long> cdcStats2) {
- final Map<String, Long> cdcStats;
- if (cdcStats1 != null && cdcStats2 != null) {
- cdcStats = new HashMap<>();
- cdcStats.putAll(cdcStats1);
- cdcStats.putAll(cdcStats2);
- } else if (cdcStats1 == null) {
- cdcStats = cdcStats2;
+ private static Map<String, Long> getMergedStringLongMap(Map<String, Long>
map1, Map<String, Long> map2) {
Review Comment:
revert.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -160,10 +160,31 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
val (readSchema, readFilters) =
getSchemaAndFiltersForRead(parquetReadStructType, hasRowIndexField)
if (FSUtils.isLogFile(filePath)) {
// NOTE: now only primary key based filtering is supported for log files
+ // Position-based merging pairs log records with the RECORD_POSITIONS
bitmap by index (see
+ // PositionBasedFileGroupRecordBuffer), which requires the record stream
to contain every
+ // record of the block. Pushing filters into the scan can physically
drop records (e.g. with
+ // parquet record-level filtering) and misalign that pairing, so push no
filters in that case.
+ val logReadFilters = if (getShouldMergeUseRecordPosition)
Seq.empty[Filter] else readFilters
// Variant alignment happens later via getLogBlockRecordProjection in
the merge buffer.
- new HoodieSparkFileReaderFactory(storage).newParquetFileReader(filePath)
-
.asInstanceOf[HoodieSparkParquetReader].getUnsafeRowIterator(requiredSchema,
readFilters.asJava)
- .asInstanceOf[ClosableIterator[InternalRow]]
+ // Log files reach this method either as an inline parquet data block or
as a native log file.
+ // Inline paths do not carry the parquet extension, so resolve them by
the log block contract.
+ val fileFormat = if
(FSUtils.matchNativeLogFile(filePath.getName).isEmpty) {
Review Comment:
add isInlineLogFile
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -611,7 +611,9 @@ public static Set<String>
getFilesToFetchColumnStats(List<HoodieWriteStat> parti
.flatMap(fileSlice -> Stream.concat(
Stream.of(fileSlice.getBaseFile().map(HoodieBaseFile::getFileName).orElse(null)),
fileSlice.getLogFiles().map(HoodieLogFile::getFileName)))
- .filter(e -> Objects.nonNull(e) && !filesWithColumnStats.contains(e)
&& !fileGroupIdsToReplace.contains(e))
+ .filter(e -> Objects.nonNull(e)
Review Comment:
revert.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java:
##########
@@ -206,7 +208,7 @@ public List<WriteStatus> close() {
if (isSecondaryIndexStatsStreamingWritesEnabled && !statuses.isEmpty()) {
SecondaryIndexStreamingTracker.trackSecondaryIndexStats(partitionPath,
fileId, getReadFileSlice(),
- statuses.stream().map(status ->
status.getStat().getPath()).collect(Collectors.toList()),
+
SecondaryIndexStreamingTracker.collectNewLogFilesForSecondaryIndexStats(statuses),
Review Comment:
revert.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestSecondaryIndexStreamingTracker.java:
##########
@@ -366,6 +367,35 @@ void testTrackSecondaryIndexStats_RecordKeyExtraction() {
});
}
+ @Test
Review Comment:
remove
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/SecondaryIndexStreamingTracker.java:
##########
@@ -127,6 +132,25 @@ static void trackSecondaryIndexStats(String partitionPath,
String fileId, Option
});
}
+ /**
+ * Collects the paths of the new log files written by an append handle, so
that they can be added
+ * to the file slice when generating secondary index stats.
+ *
+ * @param statuses Write statuses produced by the append handle
+ * @return De-duplicated list of new log file paths in insertion order
+ */
+ @VisibleForTesting
+ static List<String>
collectNewLogFilesForSecondaryIndexStats(List<WriteStatus> statuses) {
Review Comment:
remove.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeDeleteBlock.java:
##########
@@ -20,24 +20,30 @@
package org.apache.hudi.common.table.log.block;
import org.apache.hudi.common.engine.HoodieReaderContext;
-import org.apache.hudi.common.engine.RecordContext;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
import org.apache.hudi.common.table.read.BufferedRecord;
import org.apache.hudi.common.table.read.BufferedRecords;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.exception.HoodieIOException;
-import org.apache.hudi.exception.HoodieNotSupportedException;
+import org.apache.hudi.io.storage.HoodieFileReader;
+import org.apache.hudi.io.storage.HoodieIOFactory;
import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
+
+import static
org.apache.hudi.common.util.ConfigUtils.DEFAULT_HUDI_CONFIG_FOR_READER;
Review Comment:
rename, HoodieNativeLogDeleteBlock
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeDeleteBlock.java:
##########
@@ -46,51 +52,76 @@ public class HoodieNativeDeleteBlock extends
HoodieDeleteBlock {
private final HoodieStorage storage;
private final HoodieLogFile logFile;
- private final HoodieReaderContext<?> readerContext;
private final HoodieSchema deleteLogSchema;
private final List<String> orderingFieldNames;
+ private final String partitionPath;
+ private final Properties props;
+ private DeleteRecord[] recordsToDelete;
private List<BufferedRecord<?>> bufferedRecordsToDelete;
public HoodieNativeDeleteBlock(HoodieStorage storage,
HoodieLogFile logFile,
- HoodieReaderContext<?> readerContext,
- HoodieSchema deleteLogSchema,
List<String> orderingFieldNames,
+ String partitionPath,
+ Properties props,
Map<HeaderMetadataType, String> header,
Map<FooterMetadataType, String> footer) {
super(Option.empty(), null, true, getContentLocation(storage, logFile),
header, footer);
this.storage = storage;
this.logFile = logFile;
- this.readerContext = readerContext;
- this.deleteLogSchema = deleteLogSchema;
+ this.deleteLogSchema =
HoodieSchemas.createDeleteLogSchema(getSchemaFromHeader(), orderingFieldNames);
this.orderingFieldNames = orderingFieldNames;
+ this.partitionPath = partitionPath;
+ this.props = props == null ? new Properties() : props;
}
@Override
public DeleteRecord[] getRecordsToDelete() {
- throw new HoodieNotSupportedException("Native delete log files do not
support the legacy DeleteRecord[] API. "
- + "Use getRecordsToDelete(RecordContext) instead. Log file: " +
logFile);
+ if (recordsToDelete == null) {
+ recordsToDelete = readRecordsToDelete();
Review Comment:
todo: add issue, refactor.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FlinkRowDataReaderContext.java:
##########
@@ -98,24 +101,39 @@ public ClosableIterator<RowData> getFileRecordIterator(
HoodieSchema dataSchema,
HoodieSchema requiredSchema,
HoodieStorage storage) throws IOException {
- boolean isLogFile = FSUtils.isLogFile(filePath);
// disable schema evolution in fileReader if it's log file, since schema
evolution for log file is handled in `FileGroupRecordBuffer`
- InternalSchemaManager schemaManager = isLogFile ?
InternalSchemaManager.DISABLED : internalSchemaManager.get();
+ InternalSchemaManager schemaManager = FSUtils.isLogFile(filePath) ?
InternalSchemaManager.DISABLED : internalSchemaManager.get();
// Log files only reach this method for parquet data blocks; base files
are resolved by their extension.
// Format-specific handling lives in the readers themselves, so this
method stays format-agnostic.
- HoodieFileFormat format = isLogFile ? HoodieFileFormat.PARQUET :
HoodieFileFormat.fromFileExtension(filePath.getFileExtension());
+ boolean isInlineLogFile = FSUtils.isLogFile(filePath) &&
FSUtils.matchNativeLogFile(filePath.getName()).isEmpty();
+ HoodieFileFormat format = isInlineLogFile ? HoodieFileFormat.PARQUET :
HoodieFileFormat.fromFileExtension(filePath.getFileExtension());
HoodieRowDataFileReader reader = (HoodieRowDataFileReader)
HoodieIOFactory.getIOFactory(storage)
.getReaderFactory(HoodieRecord.HoodieRecordType.FLINK)
.getFileReader(tableConfig, filePath, format, Option.empty());
try {
- return reader.getRowDataIterator(dataSchema, requiredSchema,
schemaManager, getSafePredicates(requiredSchema));
+ ClosableIterator<RowData> rowDataIterator =
+ reader.getRowDataIterator(dataSchema, requiredSchema, schemaManager,
getSafePredicates(requiredSchema));
+ return resolveRowKind(rowDataIterator, requiredSchema);
} catch (Throwable e) {
reader.close();
throw new HoodieException("Failed to get record iterator for: " +
filePath, e);
}
}
+ private ClosableIterator<RowData> resolveRowKind(ClosableIterator<RowData>
rowDataIterator, HoodieSchema requiredSchema) {
Review Comment:
todo: inline parquet data block bug.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeDataBlock.java:
##########
@@ -95,11 +94,7 @@ protected <T> ClosableIterator<T>
readRecordsFromBlockPayload(HoodieReaderContex
@Override
protected <T> ClosableIterator<T> lookupEngineRecords(HoodieReaderContext<T>
readerContext, List<String> keys, boolean fullKey) throws IOException {
- return FilteringEngineRecordIterator.getInstance(
- readRecordsFromBlockPayload(readerContext),
- new HashSet<>(keys),
- fullKey,
- record ->
Option.ofNullable(readerContext.getRecordContext().getRecordKey(record,
readerSchema)));
+ return readerContext.lookupRecords(logFile.getPath(), fileFormat,
readerSchema, storage, keys, fullKey);
Review Comment:
rename, HoodieNativeLogDataBlock
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
