prakashgyank46-commits opened a new issue, #20327:
URL: https://github.com/apache/hudi/issues/20327
### Bug Description
**What happened:**
Log compaction changes `_hoodie_commit_time` on unchanged MOR records to the
log-compaction instant. On Hudi **1.2.1**, an incremental query restricted to
that service-only window initially returns zero rows, but returns an unchanged
row after a later regular compaction and cleaning force incremental full-table
fallback. This causes duplicate downstream processing even though the window
contains no logical writes.
Also reproduced on **1.1.0**, where the duplication occurs immediately after
log compaction as well. The 1.2.1 improvement before cleaning does not fully
resolve the problem.
**What you expected:**
Log compaction should preserve original record commit metadata. A window
containing only log compaction should not produce changed rows, including when
the incremental reader falls back after cleaning. A later normal write should
return only that genuinely changed record.
**Steps to reproduce:**
1. Create a fresh MOR table with fields `id`, `ts`, `value`, `part`;
bulk-insert `a` and `b` in partition `p`.
2. Update both records in three separate commits. Their final values are
`a3` and `b3` with `ts=3`. Read and consume these changes.
3. Explicitly schedule, execute and commit log compaction with block
threshold 3. No logical write occurs in this service. Save its requested time L
and completion time C.
4. Query incremental `(L, C]`. L is after all data-write completions and
before service completion, so this window contains only the log-compaction
commit. On 1.2.1 it returns zero rows; on 1.1.0 it returns both old rows.
Snapshot metadata now reports `_hoodie_commit_time=L` on both records, although
values are unchanged.
5. Upsert **only a** to `a4`, `ts=4`, in a later commit.
6. Explicitly schedule, execute and commit regular compaction.
7. Run the cleaner with `KEEP_LATEST_FILE_VERSIONS`, retaining one file
version. Verify the original base file has been removed and the latest
compacted base file remains.
8. Repeat the **same** incremental `(L, C]` query with the default
full-table fallback. Both 1.1.0 and **1.2.1 return b3 with
`_hoodie_commit_time=L`**, even though b was never modified in that window. The
last real write to b predates L.
The cleaner uses a supported policy through `SparkRDDWriteClient.clean()`;
no files/timeline entries are manually deleted or modified. The intentionally
short retention makes the delayed-consumer/full-table-fallback case easy to
reproduce.
**Results:**
| Window | 1.1.0 | 1.2.1 |
|---|---|---|
| Log compaction only, immediately | 2 duplicate rows | 0 rows |
| Log compaction + later update to a | a4 + duplicate b3 | a4 only |
| Later write only | a4 only | a4 only |
| Regular compaction only | 0 rows | 0 rows |
| Historical log-compaction window after regular compaction, before clean |
2 duplicate rows | 0 rows |
| Same historical window after clean | **duplicate b3** | **duplicate b3** |
Results match with CDC disabled and enabled (`data_before_after`), and with
`hoodie.file.group.reader.enabled` omitted, true, or false. Additionally
reproduced using 1.2.1 writing a fresh **table-version-8** table, as well as
its default table version 9. This is not an in-place upgrade test.
There were **44 incremental queries per configuration**, 132 total across
1.1.0, 1.2.1/v9, and 1.2.1/v8. Snapshot business values remained unchanged by
both compaction services. No data loss or production-scale performance claim is
made.
**Related metadata observation:** The service remains a completed
`deltacommit` with `operation=LOG_COMPACT` and nonzero `numWrites`. Consumers
that blindly sum those physical-write statistics can also overcount logical
input. That integration concern is distinct from the incorrect rows returned by
the incremental reader.
Duplicate searches for log compaction/incremental/clean, log
compaction/_hoodie_commit_time, and logcompaction/duplicate found no matching
current report. #20325 concerns a CDC scheduling NPE; this report concerns
successful log compaction followed by incorrect incremental output.
### Environment
**Hudi versions:** unmodified Spark 3.5/Scala 2.12 bundles, 1.1.0 and latest
release 1.2.1. Fresh table v8 for 1.1.0; 1.2.1 tested with both default v9 and
explicitly configured v8.
**Query engine:** Spark 3.5.6, Scala 2.12, `local[2]`.
**Runtime:** OpenJDK 11.0.32.1, macOS 26.6/aarch64, local filesystem, 3 GB
JVM heap; synthetic two-row data, no external catalog or cluster.
**Relevant configs:**
```properties
hoodie.datasource.write.table.type=MERGE_ON_READ
hoodie.datasource.write.recordkey.field=id
hoodie.datasource.write.precombine.field=ts
hoodie.datasource.write.partitionpath.field=part
hoodie.table.cdc.enabled=false
hoodie.metadata.enable=false
hoodie.compact.inline=false
hoodie.compact.schedule.inline=false
hoodie.compact.inline.max.delta.commits=1
hoodie.clean.automatic=false
hoodie.archive.automatic=false
hoodie.log.compaction.enable=false
hoodie.log.compaction.blocks.threshold=3
hoodie.embed.timeline.server=false
hoodie.write.markers.type=DIRECT
```
CDC-on runs use `hoodie.table.cdc.enabled=true` and
`hoodie.table.cdc.supplemental.logging.mode=data_before_after`. All shuffle
parallelism values are 1. Services are explicitly scheduled/executed/committed
through the write client. The final cleaner call uses
`hoodie.cleaner.policy=KEEP_LATEST_FILE_VERSIONS` and
`hoodie.cleaner.fileversions.retained=1`. Incremental query format and
full-table fallback use defaults.
1.2.1 bundle SHA-1 verified against Maven Central:
`1120807b30891d6e32e4c71512b8f4fbfa2a8d50`.
### Logs and Stack Trace
This is incorrect output, not a thrown exception. Selected 1.2.1 output:
```text
QUERY label=cdc_off_data_before reader=default begin=000
end=20261014110623942 count=2
rows=[{"id":"a","ts":3,"value":"a3","part":"p","_hoodie_commit_time":"20261014110623543"},
{"id":"b","ts":3,"value":"b3","part":"p","_hoodie_commit_time":"20261014110623543"}]
SNAPSHOT_METADATA_AFTER=[{"id":"a","_hoodie_commit_time":"20261014110624974"},
{"id":"b","_hoodie_commit_time":"20261014110624974"}]
TIMELINE requested=20261014110624974 completion=20261014110625352
action=deltacommit operation=LOG_COMPACT numWrites=2 numUpdates=2
QUERY label=cdc_off_logcompaction_only reader=default
begin=20261014110624974 end=20261014110625352 count=0 rows=[]
QUERY label=cdc_off_checkpoint_boundary reader=default
begin=20261014110623942 end=20261014110625352 count=0 rows=[]
QUERY label=cdc_off_new_write_only reader=default begin=20261014110626675
end=20261014110627053 count=1
rows=[{"id":"a","ts":4,"value":"a4","part":"p","_hoodie_commit_time":"20261014110626675"}]
QUERY label=cdc_off_logcompaction_plus_write reader=default
begin=20261014110624974 end=20261014110627053 count=1
rows=[{"id":"a","ts":4,"value":"a4","part":"p","_hoodie_commit_time":"20261014110626675"}]
QUERY label=cdc_off_regular_compaction_only reader=default
begin=20261014110627951 end=20261014110628270 count=0 rows=[]
QUERY label=cdc_off_historical_after_regular reader=default
begin=20261014110624974 end=20261014110625352 count=0 rows=[]
QUERY label=cdc_off_historical_after_clean reader=default
begin=20261014110624974 end=20261014110625352 count=1
rows=[{"id":"b","ts":3,"value":"b3","part":"p","_hoodie_commit_time":"20261014110624974"}]
SNAPSHOT_METADATA_AFTER=[{"id":"a","_hoodie_commit_time":"20261014110632061"},
{"id":"b","_hoodie_commit_time":"20261014110632061"}]
TIMELINE requested=20261014110632061 completion=20261014110632394
action=deltacommit operation=LOG_COMPACT numWrites=2 numUpdates=2
```
**Source analysis (not a patched/tested fix):**
In release-1.2.1, [FileGroupReaderBasedAppendHandle's
constructor](https://github.com/apache/hudi/blob/release-1.2.1/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedAppendHandle.java#L67)
calls the ordinary `HoodieAppendHandle` constructor. That path does not set
the private `isLogCompaction` flag, whereas the older dedicated log-compaction
constructor does.
[populateMetadataFields](https://github.com/apache/hudi/blob/release-1.2.1/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java#L327)
therefore replaces record commit time/sequence metadata with the service
instant. The observed snapshot metadata and later base-file fallback result are
consistent with this constructor path. Please cover both commit-time
preservation and delayed incremental reads in a fix.
### Standalone reproducer
Save the source below as `LogCompactionIncrementalRepro.java`. With Java 11
and `REPRO_CP` containing Spark SQL + Spark Hive 3.5.6 runtime dependencies,
commons-logging 1.2, protobuf-java 3.25.5 and exactly one matching Hudi Spark
3.5/Scala 2.12 bundle:
```sh
mkdir -p evidence classes
javac -cp "$REPRO_CP" -d classes LogCompactionIncrementalRepro.java
SPARK_LOCAL_IP=127.0.0.1 java -Xmx3g -cp "classes:$REPRO_CP"
LogCompactionIncrementalRepro
# For the 1.2.1 table-version-8 variant, add -Drepro.table.version=8 before
-cp.
```
The harness creates disposable local tables under `evidence/` and cleans
only those tables. It prints observed query rows, service metadata and query
plans; it checks normal-write and regular-compaction controls and
business-value preservation. `TEST_COMPLETED` indicates the harness completed;
a nonzero `duplicate_windows` count records reproduced failures rather than a
passing correctness test. External checks verified the counts shown in the
result table for every reader/CDC combination.
<details>
<summary>Complete Java reproducer</summary>
```java
import java.nio.file.*;
import java.util.*;
import java.util.stream.*;
import org.apache.spark.sql.*;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.TableSchemaResolver;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.table.action.HoodieWriteMetadata;
public class LogCompactionIncrementalRepro {
static SparkSession spark;
static JavaSparkContext jsc;
static Path root;
static int duplicates;
static int queries;
public static void main(String[] args) throws Exception {
TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
root=Files.createTempDirectory(Paths.get("evidence"),"tables-").toAbsolutePath();
System.out.println("EVIDENCE_ROOT="+root);
System.out.println("HUDI_RUNTIME="+SparkRDDWriteClient.class.getProtectionDomain().getCodeSource().getLocation());
spark=SparkSession.builder().master("local[2]").appName("hudi-logcompaction-incremental-repro")
.config("spark.driver.host","127.0.0.1").config("spark.driver.bindAddress","127.0.0.1")
.config("spark.ui.enabled","false").config("spark.sql.shuffle.partitions","1")
.config("spark.default.parallelism","1").config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
.config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog","org.apache.spark.sql.hudi.catalog.HoodieCatalog")
.config("spark.sql.adaptive.enabled","false").getOrCreate();
spark.sparkContext().setLogLevel("ERROR");
jsc=JavaSparkContext.fromSparkContext(spark.sparkContext());
try { runCase(false); runCase(true); System.out.println("TEST_COMPLETED
queries="+queries+" duplicate_windows="+duplicates); }
finally {spark.stop();}
}
static String schema(String path) throws Exception {
TableSchemaResolver resolver=new TableSchemaResolver(meta(path));
try {return
resolver.getClass().getMethod("getTableSchema").invoke(resolver).toString();}
catch (NoSuchMethodException e) {return
resolver.getClass().getMethod("getTableAvroSchema").invoke(resolver).toString();}
}
static Map<String,String> options(String name, boolean cdc) {
Map<String,String> p = new HashMap<>();
p.put("hoodie.table.name", name);
if (System.getProperty("repro.table.version") != null)
p.put("hoodie.write.table.version",System.getProperty("repro.table.version"));
p.put("hoodie.datasource.write.table.type", "MERGE_ON_READ");
p.put("hoodie.datasource.write.recordkey.field", "id");
p.put("hoodie.datasource.write.precombine.field", "ts");
p.put("hoodie.datasource.write.partitionpath.field", "part");
p.put("hoodie.table.cdc.enabled", String.valueOf(cdc));
p.put("hoodie.table.cdc.supplemental.logging.mode", "data_before_after");
p.put("hoodie.metadata.enable", "false");
p.put("hoodie.compact.inline", "false");
p.put("hoodie.compact.schedule.inline", "false");
p.put("hoodie.compact.inline.max.delta.commits", "1");
p.put("hoodie.clean.automatic", "false");
p.put("hoodie.archive.automatic", "false");
p.put("hoodie.log.compaction.enable", "false");
p.put("hoodie.log.compaction.blocks.threshold", "3");
p.put("hoodie.embed.timeline.server", "false");
p.put("hoodie.write.markers.type", "DIRECT");
p.put("hoodie.insert.shuffle.parallelism", "1");
p.put("hoodie.upsert.shuffle.parallelism", "1");
p.put("hoodie.bulkinsert.shuffle.parallelism", "1");
p.put("hoodie.delete.shuffle.parallelism", "1");
return p;
}
static void write(String path, Map<String,String> opts, int n) {
Dataset<Row> df = spark.read().json(spark.createDataset(Arrays.asList(
"{\"id\":\"a\",\"ts\":" + n + ",\"value\":\"a" + n +
"\",\"part\":\"p\"}",
"{\"id\":\"b\",\"ts\":" + n + ",\"value\":\"b" + n +
"\",\"part\":\"p\"}"), Encoders.STRING()));
df.coalesce(1).write().format("hudi").options(opts)
.option("hoodie.datasource.write.operation", n == 0 ? "bulk_insert"
: "upsert")
.mode(SaveMode.Append).save(path);
}
static HoodieTableMetaClient meta(String path) {
return
HoodieTableMetaClient.builder().setConf(HadoopFSUtils.getStorageConf(jsc.hadoopConfiguration()))
.setBasePath(path).setLoadActiveTimelineOnLoad(true).build();
}
static SparkRDDWriteClient client(String path, Map<String,String> opts)
throws Exception {
Properties p = new Properties(); p.putAll(opts);
HoodieWriteConfig cfg =
HoodieWriteConfig.newBuilder().withPath(path).forTable(opts.get("hoodie.table.name"))
.withSchema(schema(path)).withProps(p).build();
return new SparkRDDWriteClient(new HoodieSparkEngineContext(jsc), cfg);
}
static List<String> snapshot(String path) {
return
spark.read().format("hudi").load(path).select("id","ts","value","part").orderBy("id").toJSON().collectAsList();
}
static List<String> cdcRows(String path) {
List<String> rows = new ArrayList<>(spark.read().format("hudi")
.option("hoodie.datasource.query.type", "incremental")
.option("hoodie.datasource.query.incremental.format", "cdc")
.option("hoodie.datasource.read.begin.instanttime", "000")
.load(path).toJSON().collectAsList());
Collections.sort(rows);
return rows;
}
static HoodieInstant latest(String path) {
return
meta(path).getActiveTimeline().getCommitsTimeline().filterCompletedInstants().lastInstant().get();
}
static void timeline(String path) throws Exception {
HoodieTimeline
t=meta(path).getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
for(HoodieInstant
i:t.getInstantsAsStream().collect(Collectors.toList())) {
HoodieCommitMetadata m=t.readCommitMetadata(i);
System.out.println("TIMELINE requested="+i.requestedTime()+"
completion="+i.getCompletionTime()+" action="+i.getAction()+"
operation="+m.getOperationType()+" numWrites="+m.fetchTotalRecordsWritten()+"
numUpdates="+m.fetchTotalUpdateRecordsWritten());
}
}
static List<String> incremental(String path,String begin,String end,String
reader,String label) throws Exception {
DataFrameReader
r=spark.read().format("hudi").option("hoodie.datasource.query.type","incremental")
.option("hoodie.datasource.read.begin.instanttime",begin).option("hoodie.datasource.read.end.instanttime",end);
if(!reader.equals("default"))
r=r.option("hoodie.file.group.reader.enabled",reader);
Dataset<Row> df=r.load(path);
List<String>
rows=df.select("id","ts","value","part","_hoodie_commit_time").orderBy("id").toJSON().collectAsList();
queries++;
System.out.println("QUERY label="+label+" reader="+reader+"
begin="+begin+" end="+end+" count="+rows.size()+" rows="+rows);
Files.writeString(root.resolve(label+"-"+reader+"-plan.txt"),df.queryExecution().toString());
return rows;
}
static void runCase(boolean cdc) throws Exception {
String name=cdc?"cdc_on":"cdc_off";
String path=root.resolve(name).toString();
Map<String,String> opts=options(name,cdc);
write(path,opts,0);
for(int n=1;n<=3;n++)write(path,opts,n);
HoodieInstant data=latest(path);
List<String> before=snapshot(path);
List<String>
dataRows=incremental(path,"000",data.getCompletionTime(),"default",name+"_data_before");
if(dataRows.size()!=2)throw new AssertionError("Expected 2 original
rows");
try(SparkRDDWriteClient c=client(path,opts)) {
Option<String> i=c.scheduleLogCompaction(Option.empty());
if(!i.isPresent())throw new AssertionError("Log compaction not
scheduled");
HoodieWriteMetadata result=c.logCompact(i.get());
c.commitLogCompaction(i.get(),result,Option.empty());
}
HoodieInstant lc=latest(path);
if(lc.requestedTime().equals(data.requestedTime()))throw new
AssertionError("Missing completed log compaction");
if(!before.equals(snapshot(path)))throw new AssertionError("Snapshot
changed by log compaction");
System.out.println("SNAPSHOT_UNCHANGED case="+name);
System.out.println("SNAPSHOT_METADATA_AFTER="+spark.read().format("hudi").load(path).select("id","_hoodie_commit_time").orderBy("id").toJSON().collectAsList());
timeline(path);
for(String reader:Arrays.asList("default","true","false")) {
// Start at the service request: after all input writes completed,
before service completion.
List<String>
rows=incremental(path,lc.requestedTime(),lc.getCompletionTime(),reader,name+"_logcompaction_only");
if(!rows.isEmpty())duplicates++;
incremental(path,data.getCompletionTime(),lc.getCompletionTime(),reader,name+"_checkpoint_boundary");
}
Dataset<Row>
update=spark.read().json(spark.createDataset(Collections.singletonList("{\"id\":\"a\",\"ts\":4,\"value\":\"a4\",\"part\":\"p\"}"),Encoders.STRING()));
update.coalesce(1).write().format("hudi").options(opts).option("hoodie.datasource.write.operation","upsert").mode(SaveMode.Append).save(path);
HoodieInstant next=latest(path);
for(String reader:Arrays.asList("default","true","false")) {
List<String>
rows=incremental(path,next.requestedTime(),next.getCompletionTime(),reader,name+"_new_write_only");
if(rows.size()!=1 || !rows.get(0).contains("a4"))throw new
AssertionError("New write not read correctly: "+rows);
List<String>
mixed=incremental(path,lc.requestedTime(),next.getCompletionTime(),reader,name+"_logcompaction_plus_write");
if(mixed.size()>1)duplicates++;
}
List<String> beforeRegular=snapshot(path);
try(SparkRDDWriteClient c=client(path,opts)) {
Option<String> i=c.scheduleCompaction(Option.empty());
if(!i.isPresent())throw new AssertionError("Regular compaction not
scheduled");
HoodieWriteMetadata result=c.compact(i.get());
c.commitCompaction(i.get(),result,Option.empty());
}
HoodieInstant regular=latest(path);
if(!beforeRegular.equals(snapshot(path)))throw new
AssertionError("Snapshot changed by regular compaction");
for(String reader:Arrays.asList("default","true","false")) {
List<String>
rows=incremental(path,regular.requestedTime(),regular.getCompletionTime(),reader,name+"_regular_compaction_only");
if(!rows.isEmpty())throw new AssertionError("Regular compaction
control reemits: "+rows);
}
for(String reader:Arrays.asList("default","true","false")) {
List<String>
historical=incremental(path,lc.requestedTime(),lc.getCompletionTime(),reader,name+"_historical_after_regular");
if(!historical.isEmpty())duplicates++;
}
opts.put("hoodie.cleaner.policy","KEEP_LATEST_FILE_VERSIONS");
opts.put("hoodie.cleaner.fileversions.retained","1");
try(SparkRDDWriteClient c=client(path,opts)) { c.clean(); }
System.out.println("AFTER_CLEAN_FILES="+Files.walk(Paths.get(path)).filter(p->p.getFileName().toString().endsWith(".parquet")).collect(Collectors.toList()));
for(String reader:Arrays.asList("default","true","false")) {
List<String>
historical=incremental(path,lc.requestedTime(),lc.getCompletionTime(),reader,name+"_historical_after_clean");
if(!historical.isEmpty())duplicates++;
}
System.out.println("CASE_COMPLETED="+name);
}
}
```
</details>
--
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]
