apache / apache/incubator-xtable
Parquet source: record_count is never populated, so optimized count(*) returns 0 on non-empty tables
- Dominant language
- Java
- Stars
- 1.2k
- Forks
- 212
- Avg merge
- 4d 9h
- Merged PRs (30d)
- 16
Description
### Search before asking
- [x] I had searched in the [issues](https://github.com/apache/incubator-xtable/issues?q=is%3Aissue) and found no similar issues.
### Please describe the bug 🐞
### Describe the bug
Converting a plain Parquet source to Iceberg (or Delta) reports success, and the target references all source files, but every data file is registered with `record_count = 0`. Any engine that answers aggregates from table metadata then returns **0 rows for a non-empty table, with no error**: Spark's optimized `count(*)`, Iceberg's `.files` metadata, and Delta's `numRecords` all show it. A forced scan (e.g. `count(distinct id)`) still returns the real rows, which makes the wrongness silent and easy to miss.
### Root cause
Both `InternalDataFile` construction sites in `ParquetConversionSource` (full sync and incremental) populate path, size, partition values, modification time, and column stats, but never call `.recordCount(...)`: https://github.com/apache/incubator-xtable/blob/e8f22867a6cb297f58b248150220fce71188d16a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java#L88-L152
`InternalFile.recordCount` is a primitive `long` with no builder default, so it stays 0: https://github.com/apache/incubator-xtable/blob/e8f22867a6cb297f58b248150220fce71188d16a/xtable-api/src/main/java/org/apache/xtable/model/storage/InternalFile.java#L39-L55
Both targets serialize the zero faithfully. The row count is already available at the call site: the Parquet footer is read for `columnStats` in the same expression, and `ParquetStatsExtractor` correctly computes per-column `numValues` from it (which is why `value_counts` come out right while `record_count` is 0). Summing `BlockMetaData.getRowCount()` over the footer's blocks would fix it.
Why existing tests pass: `ITParquetConversionSource` materializes rows to JSON and compares lists, which forces file scans; it never asserts Iceberg `record_count`, Delta `numRecords`, or an optimized `count(*)`: https://github.com/apache/incubator-xtable/blob/e8f22867a6cb297f58b248150220fce71188d16a/xtable-core/src/test/java/org/apache/xtable/parquet/ITParquetConversionSource.java#L362-L435
### To Reproduce
Environment: JDK 17, `pip install pyspark==3.5.5`, jars from Maven Central: `xtable-spark-runtime_2.12-0.4.0-incubating.jar`, `iceberg-spark-runtime-3.5_2.12-1.9.2.jar`. Local filesystem. Reproduction scripts are included at the bottom of this issue.
```bash
# 1. a plain partitioned Parquet table, 1,000 rows
spark-submit gen_parquet.py /tmp/orders_parquet
# 2. sync: reports success, exit 0
spark-submit --master 'local[2]' \
--class org.apache.xtable.spark.XTableSparkSync \
--jars iceberg-spark-runtime-3.5_2.12-1.9.2.jar \
xtable-spark-runtime_2.12-0.4.0-incubating.jar \
--basepath /tmp/orders_parquet --sourceformat PARQUET --targets ICEBERG
# log: "Sync is successful for the following formats ICEBERG"
# 3. inspect the target
spark-submit --jars iceberg-spark-runtime-3.5_2.12-1.9.2.jar \
inspect_iceberg.py /tmp/orders_parquet
```
Observed:
```text
optimized count(*) : 0 <- wrong, silently
forced distinct order_id: 1000 <- rows are physically present
record_count = 0 for every file; value_counts are correct
```
### Expected behavior
`record_count` reflects the actual row count of each registered file, and optimized `count(*)` on the target returns the source row count.
### Affected versions
Reproduced against the published `0.4.0-incubating` artifact (SHA-1 `ebbb39fdc6862eaa82900126cc75cc4fea94153e`). The same code is present on current `main` (`ParquetConversionSource` unchanged in the relevant methods; re-checked 2026-09-04). Also reproduced with a Delta target: `numRecords=0` in every add action, and optimized `count(*)` = 0 there as well.
#### Reproduction scripts
gen_parquet.py
```python
import sys
from pyspark.sql import SparkSession, functions as F
path = sys.argv[1]
spark = SparkSession.builder.getOrCreate()
(spark.range(0, 1000).withColumn("order_id", F.col("id"))
.withColumn("order_date", F.lit("2026-08-01")).drop("id").repartition(2)
.write.mode("overwrite").partitionBy("order_date").parquet(path))
print("wrote 1000 rows ->", path)
```
inspect_iceberg.py
```python
import sys
from pyspark.sql import SparkSession, functions as F
path = sys.argv[1]
spark = SparkSession.builder.getOrCreate()
df = spark.read.format("iceberg").load(path)
print("optimized count(*) :", df.count())
print("forced distinct order_id:", df.select(F.countDistinct("order_id")).collect()[0][0])
print("columns :", df.columns)
spark.read.format("iceberg").load(path + "#files") \
.select("file_path", "record_count", "value_counts").show(truncate=False)
if "--partitions" in sys.argv:
try:
spark.read.format("iceberg").load(path + "#partitions").show(truncate=False)
except Exception as e:
print("partitions metadata unavailable:", type(e).__name__)
```
### Are you willing to submit PR?
- [x] I am willing to submit a PR!
- [ ] I am willing to submit a PR but need help getting started!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.