Skip to content

[HUDI-3513] Make sure Column Stats does not fail in case it fails to load previous Index Table state #5015

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nsivabalan merged 2 commits into from
Mar 11, 2022
Merged

[HUDI-3513] Make sure Column Stats does not fail in case it fails to load previous Index Table state #5015

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
16c497f
Make sure Column Stats does not fail in case it fails to load previou…
Mar 11, 2022
4b26887
Cleaning up unused imports
Mar 11, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 34 additions & 15 deletions ...-spark-client/src/main/java/org/apache/hudi/index/columnstats/ColumnStatsIndexHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.hudi.common.model.HoodieColumnRangeMetadata;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.util.BaseFileUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ParquetUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
Expand Down Expand Up @@ -304,21 +305,28 @@ public static void updateColumnStatsIndexFor(
if (validIndexTables.isEmpty()) {
finalColStatsIndexDf = newColStatsIndexDf;
} else {
// NOTE: That Parquet schema might deviate from the original table schema (for ex,
// by upcasting "short" to "integer" types, etc), and hence we need to re-adjust it
// prior to merging, since merging might fail otherwise due to schemas incompatibility
finalColStatsIndexDf =
tryMergeMostRecentIndexTableInto(
sparkSession,
newColStatsIndexDf,
// Load current most recent col-stats-index table
sparkSession.read().load(
new Path(indexFolderPath, validIndexTables.get(validIndexTables.size() - 1)).toString()
)
);

// Clean up all index tables (after creation of the new index)
tablesToCleanup.addAll(validIndexTables);
Path latestIndexTablePath = new Path(indexFolderPath, validIndexTables.get(validIndexTables.size() - 1));

Option<Dataset<Row>> existingIndexTableOpt =
tryLoadExistingIndexTable(sparkSession, latestIndexTablePath);

if (!existingIndexTableOpt.isPresent()) {
finalColStatsIndexDf = newColStatsIndexDf;
} else {
// NOTE: That Parquet schema might deviate from the original table schema (for ex,
// by upcasting "short" to "integer" types, etc), and hence we need to re-adjust it
// prior to merging, since merging might fail otherwise due to schemas incompatibility
finalColStatsIndexDf =
tryMergeMostRecentIndexTableInto(
sparkSession,
newColStatsIndexDf,
// Load current most recent col-stats-index table
existingIndexTableOpt.get()
);

// Clean up all index tables (after creation of the new index)
tablesToCleanup.addAll(validIndexTables);
}
}

// Persist new col-stats-index table
Expand Down Expand Up @@ -349,6 +357,17 @@ public static void updateColumnStatsIndexFor(
}
}

@Nonnull
private static Option<Dataset<Row>> tryLoadExistingIndexTable(@Nonnull SparkSession sparkSession, @Nonnull Path indexTablePath) {
try {
Dataset<Row> indexTableDataset = sparkSession.read().load(indexTablePath.toUri().toString());
return Option.of(indexTableDataset);
} catch (Exception e) {
LOG.error(String.format("Failed to load existing Column Stats index table from (%s)", indexTablePath), e);
return Option.empty();
}
}

@Nonnull
private static Dataset<Row> tryMergeMostRecentIndexTableInto(
@Nonnull SparkSession sparkSession,
Expand Down