Skip to content

GH-3795: Preserve physical leaf identity for duplicate primitive roots #3796

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

Open
ErikBPF wants to merge 1 commit into
base: master
Choose a base branch
from
+450 −7
Open

GH-3795: Preserve physical leaf identity for duplicate primitive roots #3796

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
52 changes: 48 additions & 4 deletions ...et-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1870,6 +1870,21 @@ public ColumnChunkMetaData buildColumnChunkMetaData(
fromParquetStatistics(metaData.geospatial_statistics, type));
}

private static void collectPhysicalLeaves(
GroupType group, List<String> parent, List<PrimitiveType> leaves, List<List<String>> paths) {
// Row-group columns follow schema leaf order; names need not identify a unique leaf.
for (org.apache.parquet.schema.Type field : group.getFields()) {
List<String> path = new ArrayList<>(parent);
path.add(field.getName());
if (field.isPrimitive()) {
leaves.add(field.asPrimitiveType());
paths.add(path);
} else {
collectPhysicalLeaves(field.asGroupType(), path, leaves, paths);
}
}
}

public ParquetMetadata fromParquetMetadata(FileMetaData parquetMetadata) throws IOException {
return fromParquetMetadata(parquetMetadata, null, false);
}
Expand All @@ -1887,6 +1902,9 @@ public ParquetMetadata fromParquetMetadata(
Map<RowGroup, Long> rowGroupToRowIndexOffsetMap)
throws IOException {
MessageType messageType = fromParquetSchema(parquetMetadata.getSchema(), parquetMetadata.getColumn_orders());
List<PrimitiveType> physicalLeaves = new ArrayList<>();
List<List<String>> physicalPaths = new ArrayList<>();
collectPhysicalLeaves(messageType, Collections.emptyList(), physicalLeaves, physicalPaths);
org.apache.parquet.hadoop.metadata.FileMetaData fileMetaData =
buildFileMetaData(parquetMetadata, messageType, encryptedFooter, fileDecryptor);
String createdBy = fileMetaData.getCreatedBy();
Expand All @@ -1912,7 +1930,10 @@ public ParquetMetadata fromParquetMetadata(
blockMetaData.setOrdinal(rowGroup.getOrdinal());
}
List<ColumnChunk> columns = rowGroup.getColumns();
String filePath = columns.get(0).getFile_path();
if (columns == null || columns.size() != physicalLeaves.size()) {
throw new ParquetDecodingException("Row-group column count does not match schema leaves");
}
String filePath = columns.isEmpty() ? null : columns.get(0).getFile_path();
int columnOrdinal = -1;
for (ColumnChunk columnChunk : columns) {
columnOrdinal++;
Expand All @@ -1923,6 +1944,13 @@ public ParquetMetadata fromParquetMetadata(
}
ColumnMetaData metaData = columnChunk.meta_data;
ColumnCryptoMetaData cryptoMetaData = columnChunk.getCrypto_metadata();
if (metaData != null || cryptoMetaData == null) {
validatePhysicalLeaf(
metaData,
physicalPaths.get(columnOrdinal),
physicalLeaves.get(columnOrdinal),
columnOrdinal);
}
ColumnChunkMetaData column = null;
ColumnPath columnPath = null;
boolean lazyMetadataDecryption = false;
Expand Down Expand Up @@ -1964,6 +1992,11 @@ public ParquetMetadata fromParquetMetadata(
columnPath + ". Failed to decrypt column metadata", e);
}
}
validatePhysicalLeaf(
metaData,
physicalPaths.get(columnOrdinal),
physicalLeaves.get(columnOrdinal),
columnOrdinal);
fileDecryptor.setColumnCryptoMetadata(columnPath, true, true, (byte[]) null, columnOrdinal);
} else { // Column encrypted with column key
// setColumnCryptoMetadata triggers KMS interaction, hence delayed until this column is
Expand All @@ -1973,8 +2006,7 @@ public ParquetMetadata fromParquetMetadata(
}

if (!lazyMetadataDecryption) { // full column metadata (with stats) is available
PrimitiveType primitiveType =
messageType.getType(columnPath.toArray()).asPrimitiveType();
PrimitiveType primitiveType = physicalLeaves.get(columnOrdinal);
column =
buildColumnChunkMetaData(metaData, columnPath, primitiveType, writerVersion, createdBy);
column.setRowGroupOrdinal(rowGroup.getOrdinal());
Expand All @@ -1988,13 +2020,17 @@ public ParquetMetadata fromParquetMetadata(
// Metadata will be decrypted later, if this column is accessed
EncryptionWithColumnKey columnKeyStruct = cryptoMetaData.getENCRYPTION_WITH_COLUMN_KEY();
List<String> pathList = columnKeyStruct.getPath_in_schema();
if (!physicalPaths.get(columnOrdinal).equals(pathList)) {
throw new ParquetDecodingException(
"Encrypted column path does not match schema leaf ordinal " + columnOrdinal);
}
byte[] columnKeyMetadata = columnKeyStruct.getKey_metadata();
columnPath = ColumnPath.get(pathList.toArray(new String[pathList.size()]));
byte[] encryptedMetadataBuffer = columnChunk.getEncrypted_column_metadata();
column = ColumnChunkMetaData.getWithEncryptedMetadata(
this,
columnPath,
messageType.getType(columnPath.toArray()).asPrimitiveType(),
physicalLeaves.get(columnOrdinal),
encryptedMetadataBuffer,
columnKeyMetadata,
fileDecryptor,
Expand Down Expand Up @@ -2043,6 +2079,14 @@ private static org.apache.parquet.hadoop.metadata.FileMetaData buildFileMetaData
messageType, keyValueMetaData, createdBy, encryptionType, fileDecryptor);
}

private void validatePhysicalLeaf(ColumnMetaData metadata, List<String> path, PrimitiveType type, int ordinal) {
if (metadata == null
|| !path.equals(metadata.getPath_in_schema())
|| getType(type.getPrimitiveTypeName()) != metadata.getType()) {
throw new ParquetDecodingException("Column metadata does not match schema leaf ordinal " + ordinal);
}
}

private static IndexReference toColumnIndexReference(ColumnChunk columnChunk) {
if (columnChunk.isSetColumn_index_offset() && columnChunk.isSetColumn_index_length()) {
return new IndexReference(columnChunk.getColumn_index_offset(), columnChunk.getColumn_index_length());
Expand Down
2 changes: 1 addition & 1 deletion parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnIndexStoreImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ private ColumnIndexStoreImpl(ParquetFileReader reader, BlockMetaData block, Set<
Map<ColumnPath, IndexStore> store = new HashMap<>();
for (ColumnChunkMetaData column : block.getColumns()) {
ColumnPath path = column.getPath();
if (paths.contains(path)) {
if (paths.contains(path) && (path.size() != 1 || !store.containsKey(path))) {
store.put(path, new IndexStoreImpl(column));
}
}
Expand Down
6 changes: 4 additions & 2 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1175,10 +1175,11 @@ private ColumnChunkPageReadStore internalReadRowGroup(int blockIndex) throws IOE
// prepare the list of consecutive parts to read them in one scan
List<ConsecutivePartList> allParts = new ArrayList<>();
ConsecutivePartList currentParts = null;
Set<ColumnPath> selectedPaths = new HashSet<>();
for (ColumnChunkMetaData mc : block.getColumns()) {
ColumnPath pathKey = mc.getPath();
ColumnDescriptor columnDescriptor = paths.get(pathKey);
if (columnDescriptor != null) {
if (columnDescriptor != null && (pathKey.size() != 1 || selectedPaths.add(pathKey))) {
BenchmarkCounter.incrementTotalBytes(mc.getTotalSize());
long startingPos = mc.getStartingPos();
// first part or not consecutive => new list
Expand Down Expand Up @@ -1443,10 +1444,11 @@ private ColumnChunkPageReadStore internalReadFilteredRowGroup(
ChunkListBuilder builder = new ChunkListBuilder(block.getRowCount());
List<ConsecutivePartList> allParts = new ArrayList<>();
ConsecutivePartList currentParts = null;
Set<ColumnPath> selectedPaths = new HashSet<>();
for (ColumnChunkMetaData mc : block.getColumns()) {
ColumnPath pathKey = mc.getPath();
ColumnDescriptor columnDescriptor = paths.get(pathKey);
if (columnDescriptor != null) {
if (columnDescriptor != null && (pathKey.size() != 1 || selectedPaths.add(pathKey))) {
OffsetIndex offsetIndex = ciStore.getOffsetIndex(mc.getPath());

OffsetIndex filteredOffsetIndex =
Expand Down
Loading