This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/main by this push:
new 77003d71 fix: Populate record_count for parquet source files (#929)
77003d71 is described below
commit 77003d7124b4221e9c5061cb3d743e7ccf49dcf8
Author: Arvin <[email protected]>
AuthorDate: Sat Sep 26 07:53:09 2026 +0800
fix: Populate record_count for parquet source files (#929)
* [Bug] Populate record_count for parquet source files
Fixes #926: InternalDataFile construction in ParquetConversionSource never
set recordCount, so targets registered every file with record_count = 0 and
optimized count(*) returned 0 on non-empty tables.
Sum BlockMetaData.getRowCount() from the parquet footer and set it on both
full-sync and incremental-sync file construction paths.
* Add tests for parquet source record counts
---------
Co-authored-by: Vinish Reddy <[email protected]>
---
.../xtable/parquet/ParquetConversionSource.java | 39 +++++-----
.../xtable/parquet/ParquetMetadataExtractor.java | 6 ++
.../xtable/parquet/ITParquetConversionSource.java | 12 +++
.../parquet/TestParquetMetadataExtractor.java | 90 ++++++++++++++++++++++
4 files changed, 128 insertions(+), 19 deletions(-)
diff --git
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java
index 3a505bb8..9b132404 100644
---
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java
+++
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java
@@ -118,25 +118,28 @@ public class ParquetConversionSource implements
ConversionSource<Long> {
private Stream<InternalDataFile> getInternalDataFiles(
Stream<LocatedFileStatus> parquetFiles, InternalSchema schema) {
return parquetFiles.map(
- file ->
- InternalDataFile.builder()
- .physicalPath(file.getPath().toString())
- .fileFormat(FileFormat.APACHE_PARQUET)
- .fileSizeBytes(file.getLen())
- .partitionValues(
- partitionValueExtractor.extractPartitionValues(
- partitionSpecExtractor.spec(schema),
- HudiPathUtils.getPartitionPath(new Path(basePath),
file.getPath())))
- .lastModified(file.getModificationTime())
- .columnStats(
- parquetStatsExtractor.getStatsForFile(
-
parquetMetadataExtractor.readParquetMetadata(hadoopConf, file.getPath()),
- schema))
- .build());
+ file -> {
+ ParquetMetadata parquetMetadata =
+ parquetMetadataExtractor.readParquetMetadata(hadoopConf,
file.getPath());
+ return InternalDataFile.builder()
+ .physicalPath(file.getPath().toString())
+ .fileFormat(FileFormat.APACHE_PARQUET)
+ .fileSizeBytes(file.getLen())
+
.recordCount(parquetMetadataExtractor.getRowCount(parquetMetadata))
+ .partitionValues(
+ partitionValueExtractor.extractPartitionValues(
+ partitionSpecExtractor.spec(schema),
+ HudiPathUtils.getPartitionPath(new Path(basePath),
file.getPath())))
+ .lastModified(file.getModificationTime())
+
.columnStats(parquetStatsExtractor.getStatsForFile(parquetMetadata, schema))
+ .build();
+ });
}
public InternalDataFile createInternalDataFileFromParquetFile(
FileStatus parquetFile, InternalSchema schema) {
+ ParquetMetadata parquetMetadata =
+ parquetMetadataExtractor.readParquetMetadata(hadoopConf,
parquetFile.getPath());
return InternalDataFile.builder()
.physicalPath(parquetFile.getPath().toString())
.partitionValues(
@@ -145,10 +148,8 @@ public class ParquetConversionSource implements
ConversionSource<Long> {
HudiPathUtils.getPartitionPath(new Path(basePath),
parquetFile.getPath())))
.lastModified(parquetFile.getModificationTime())
.fileSizeBytes(parquetFile.getLen())
- .columnStats(
- parquetStatsExtractor.getStatsForFile(
- parquetMetadataExtractor.readParquetMetadata(hadoopConf,
parquetFile.getPath()),
- schema))
+ .recordCount(parquetMetadataExtractor.getRowCount(parquetMetadata))
+ .columnStats(parquetStatsExtractor.getStatsForFile(parquetMetadata,
schema))
.build();
}
diff --git
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetMetadataExtractor.java
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetMetadataExtractor.java
index f022b753..70259b40 100644
---
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetMetadataExtractor.java
+++
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetMetadataExtractor.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.parquet.HadoopReadOptions;
import org.apache.parquet.ParquetReadOptions;
import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.hadoop.util.HadoopInputFile;
import org.apache.parquet.schema.MessageType;
@@ -47,6 +48,11 @@ public class ParquetMetadataExtractor {
return footer.getFileMetaData().getSchema();
}
+ /** Returns the total number of rows in the parquet file by summing row
counts of all blocks. */
+ public long getRowCount(ParquetMetadata footer) {
+ return
footer.getBlocks().stream().mapToLong(BlockMetaData::getRowCount).sum();
+ }
+
public ParquetMetadata readParquetMetadata(Configuration conf, Path
filePath) {
ParquetReadOptions options = HadoopReadOptions.builder(conf,
filePath).build();
try (ParquetFileReader fileReader =
diff --git
a/xtable-core/src/test/java/org/apache/xtable/parquet/ITParquetConversionSource.java
b/xtable-core/src/test/java/org/apache/xtable/parquet/ITParquetConversionSource.java
index be3d0b43..cededf35 100644
---
a/xtable-core/src/test/java/org/apache/xtable/parquet/ITParquetConversionSource.java
+++
b/xtable-core/src/test/java/org/apache/xtable/parquet/ITParquetConversionSource.java
@@ -537,10 +537,22 @@ public class ITParquetConversionSource {
assertNotNull(result.getReadSchema());
InternalSnapshot snapshot = conversionSource.getCurrentSnapshot();
assertNotNull(snapshot);
+ // record counts come from the parquet footers: 4 initial rows plus 3
appended rows
+ assertEquals(
+ data.size() + futureDataToSync.size(),
+ snapshot.getPartitionedDataFiles().stream()
+ .flatMap(group -> group.getDataFiles().stream())
+ .mapToLong(InternalDataFile::getRecordCount)
+ .sum());
TableChange changes = conversionSource.getTableChangeForCommit(testTime);
assertNotNull(changes);
assertFalse(changes.getFilesDiff().dataFilesAdded().isEmpty(), "Should
have found added files");
assertEquals(expectedAddedFiles, changes.getFilesDiff().dataFilesAdded());
+ assertEquals(
+ futureDataToSync.size(),
+ changes.getFilesDiff().dataFilesAdded().stream()
+ .mapToLong(InternalDataFile::getRecordCount)
+ .sum());
// The committed source identifier must be the most-recent modification
time from the same
// listing used to compute filesAdded, otherwise a file could be marked
synced without being
// included in the changes.
diff --git
a/xtable-core/src/test/java/org/apache/xtable/parquet/TestParquetMetadataExtractor.java
b/xtable-core/src/test/java/org/apache/xtable/parquet/TestParquetMetadataExtractor.java
new file mode 100644
index 00000000..cadd12da
--- /dev/null
+++
b/xtable-core/src/test/java/org/apache/xtable/parquet/TestParquetMetadataExtractor.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.parquet;
+
+import static org.apache.parquet.column.Encoding.BIT_PACKED;
+import static org.apache.parquet.column.Encoding.PLAIN;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.IOException;
+import java.util.HashMap;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetFileWriter;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.example.GroupWriteSupport;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class TestParquetMetadataExtractor {
+
+ private static final MessageType SCHEMA =
+ new MessageType("message",
Types.required(PrimitiveTypeName.INT32).named("col"));
+
+ private final Configuration conf = new Configuration();
+ private final ParquetMetadataExtractor metadataExtractor =
ParquetMetadataExtractor.getInstance();
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void getRowCountSingleRowGroup() throws IOException {
+ Path path = new Path(tempDir.resolve("single-rg.parquet").toUri());
+ GroupWriteSupport.setSchema(SCHEMA, conf);
+ SimpleGroupFactory factory = new SimpleGroupFactory(SCHEMA);
+ try (ParquetWriter<Group> writer =
ExampleParquetWriter.builder(path).withConf(conf).build()) {
+ for (int i = 0; i < 3; i++) {
+ writer.write(factory.newGroup().append("col", i));
+ }
+ }
+
+ assertEquals(
+ 3L,
metadataExtractor.getRowCount(metadataExtractor.readParquetMetadata(conf,
path)));
+ }
+
+ @Test
+ void getRowCountSumsAllRowGroups() throws IOException {
+ Path path = new Path(tempDir.resolve("multi-rg.parquet").toUri());
+ ColumnDescriptor col = SCHEMA.getColumns().get(0);
+ ParquetFileWriter writer = new ParquetFileWriter(conf, SCHEMA, path);
+ writer.start();
+ writer.startBlock(10);
+ writer.startColumn(col, 10, CompressionCodecName.UNCOMPRESSED);
+ writer.writeDataPage(10, 4, BytesInput.fromInt(0), BIT_PACKED, BIT_PACKED,
PLAIN);
+ writer.endColumn();
+ writer.endBlock();
+ writer.startBlock(5);
+ writer.startColumn(col, 5, CompressionCodecName.UNCOMPRESSED);
+ writer.writeDataPage(5, 4, BytesInput.fromInt(0), BIT_PACKED, BIT_PACKED,
PLAIN);
+ writer.endColumn();
+ writer.endBlock();
+ writer.end(new HashMap<>());
+
+ assertEquals(
+ 15L,
metadataExtractor.getRowCount(metadataExtractor.readParquetMetadata(conf,
path)));
+ }
+}
