hudi-agent commented on code in PR #12265:
URL: https://github.com/apache/hudi/pull/12265#discussion_r3686839074
##########
pom.xml:
##########
@@ -29,7 +29,7 @@
<groupId>org.apache.hudi</groupId>
<artifactId>hudi</artifactId>
<packaging>pom</packaging>
- <version>1.0.0-SNAPSHOT</version>
+ <version>1.0.0-rc1</version>
Review Comment:
🤖 This flips the project version from `1.0.0-SNAPSHOT` to `1.0.0-rc1` here
and in all 11 packaging bundle poms. That's a release-candidate bump unrelated
to adding an integration test framework — was this an accidental inclusion from
a rebase onto a release-prep branch? Leaving `-rc1` in would break snapshot
deploys and downstream artifact resolution, so it should probably be reverted
to `-SNAPSHOT`.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/pom.xml:
##########
@@ -20,7 +20,7 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>1.0.0-SNAPSHOT</version>
+ <version>1.0.0-rc1</version>
Review Comment:
🤖 This chunk flips every module (and the docker poms) from `1.0.0-SNAPSHOT`
to the release-candidate `1.0.0-rc1`, which seems unrelated to an
integration-test-framework PR. Was this intended? A stray `rc1` bump would
break the SNAPSHOT dev/deploy flow and likely came from branching off a release
tag or a bad rebase — could you confirm and revert if it's accidental?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-kafka-connect/pom.xml:
##########
@@ -19,13 +19,13 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>1.0.0-SNAPSHOT</version>
+ <version>1.0.0-rc1</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hudi-kafka-connect</artifactId>
<description>Kafka Connect Sink Connector for Hudi</description>
- <version>1.0.0-SNAPSHOT</version>
+ <version>1.0.0-rc1</version>
Review Comment:
🤖 Is this SNAPSHOT → `1.0.0-rc1` version bump intended to be part of this
PR? A fixed release-candidate version usually lives on a release branch/tag
rather than mainline, and merging it would break `-SNAPSHOT` dev/downstream
builds. Also worth confirming every module (including the root pom in the other
chunks) is bumped in lockstep — a partial bump breaks the reactor via
parent/child version mismatch.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-integration-test/src/test/scala/org/apache/hudi/integration/test/TestBatchRowSourceDataValidator.scala:
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.hudi.integration.test
+
+import org.apache.hudi.testutils.HoodieSparkClientTestBase
+
+import org.junit.jupiter.api.Test
+
+class TestBatchRowSourceDataValidator extends HoodieSparkClientTestBase {
+ @Test
Review Comment:
🤖 This test points at a hardcoded absolute path on a local machine
(`/Users/ethan/Work/tmp/...`), so it will fail anywhere else, including CI,
since that table won't exist. Looks like leftover local scaffolding — worth
parameterizing (e.g. write a temp table in the test) or removing before merge.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/pom.xml:
##########
@@ -20,12 +20,12 @@
<parent>
<artifactId>hudi</artifactId>
<groupId>org.apache.hudi</groupId>
- <version>1.0.0-SNAPSHOT</version>
+ <version>1.0.0-rc1</version>
Review Comment:
🤖 These modules are being bumped from `1.0.0-SNAPSHOT` to `1.0.0-rc1`, and
the same change shows up across ~20 poms in this PR. Was that intended as part
of the integration-test framework? Version bumps are usually done in a
dedicated release-prep commit, and mixing this in could interfere with the
SNAPSHOT build/release flow.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-integration-test/src/test/scala/org/apache/hudi/integration/test/TestBatchRowSourceDataValidator.scala:
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.hudi.integration.test
+
+import org.apache.hudi.testutils.HoodieSparkClientTestBase
+
+import org.junit.jupiter.api.Test
+
+class TestBatchRowSourceDataValidator extends HoodieSparkClientTestBase {
+ @Test
+ def testValidate() {
+ val basePath =
"/Users/ethan/Work/tmp/hudi-1.0.0-testing/2024-11-09-13-08-28-streamer-mor-upsert-async/test_table"
Review Comment:
🤖 nit: the test has a hardcoded absolute path (`/Users/ethan/Work/tmp/...`)
that won't work on any other machine — this looks like it was left in
accidentally. Could you replace it with a generated temp path or remove it if
the test isn't ready yet?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-integration-test/src/main/java/org/apache/hudi/integration/test/BatchRowSource.java:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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.hudi.integration.test;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.HoodieStorageUtils;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.sources.RowSource;
+import org.apache.hudi.utilities.streamer.StreamContext;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+public class BatchRowSource extends RowSource {
+ public static final String PROP_BATCH_ROW_SOURCE_PATH =
"benchmark.input.source.path";
+ private static final Logger LOG =
LoggerFactory.getLogger(BatchRowSource.class);
+
+ public BatchRowSource(TypedProperties props,
+ JavaSparkContext sparkContext,
+ SparkSession sparkSession,
+ SchemaProvider schemaProvider) {
+ super(props, sparkContext, sparkSession, schemaProvider);
+ }
+
+ public BatchRowSource(TypedProperties props,
+ JavaSparkContext sparkContext,
+ SparkSession sparkSession,
+ StreamContext streamContext) {
+ super(props, sparkContext, sparkSession, streamContext);
+ }
+
+ @Override
+ protected Pair<Option<Dataset<Row>>, String> fetchNextBatch(Option<String>
lastCkptStr,
+ long
sourceLimit) {
+ String sourceDataPath = props.getString(PROP_BATCH_ROW_SOURCE_PATH);
+ if (sourceDataPath == null) {
+ throw new IllegalArgumentException(PROP_BATCH_ROW_SOURCE_PATH + " is not
set.");
+ }
+ int roundNumber = lastCkptStr.isPresent() ?
Integer.parseInt(lastCkptStr.get()) : 0;
+ String path = sourceDataPath + "/" + roundNumber;
+ StoragePath pathToFetch = new StoragePath(path);
+ HoodieStorage storage = HoodieStorageUtils.getStorage(
+ pathToFetch, new
HadoopStorageConfiguration(sparkContext.hadoopConfiguration()));
+ try {
+ if (storage.exists(pathToFetch)) {
+ return Pair.of(
+ Option.of(sparkSession.read().parquet(path + "/*.parquet")),
+ Integer.toString(roundNumber + 1));
+ }
+ } catch (IOException e) {
Review Comment:
🤖 On an IOException here the source only logs a warning and then falls
through to returning an empty batch with the checkpoint unchanged, which the
streamer treats as end-of-data. A transient error (e.g. S3 throttling / listing
failure) would silently stop the benchmark early rather than retrying. Could
this rethrow, or otherwise distinguish a transient failure from 'no more
rounds'?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-integration-test/src/main/java/org/apache/hudi/common/model/ValueSumPayload.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.hudi.common.model;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.IOException;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class ValueSumPayload extends DefaultHoodieRecordPayload {
+ private static final String INDIVIDUAL_VALUE_FIELD = "long_value";
+ private static final String TOTAL_VALUE_FIELD = "sum_long_value";
+ private AtomicBoolean isDeleteComputed = new AtomicBoolean(false);
Review Comment:
🤖 nit: `isDeleteComputed` and `isDefaultRecordPayloadDeleted` are instance
fields used to cache a computation across calls, but `ValueSumPayload`
instances are typically short-lived and single-use — could you clarify with a
comment why these need to be stateful fields rather than local variables inside
`combineAndGetUpdateValue`? The `AtomicBoolean` in particular implies
thread-safety concerns that aren't obvious here.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-integration-test/src/main/java/org/apache/hudi/common/model/ValueSumPayload.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.hudi.common.model;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.IOException;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class ValueSumPayload extends DefaultHoodieRecordPayload {
+ private static final String INDIVIDUAL_VALUE_FIELD = "long_value";
+ private static final String TOTAL_VALUE_FIELD = "sum_long_value";
+ private AtomicBoolean isDeleteComputed = new AtomicBoolean(false);
+ private boolean isDefaultRecordPayloadDeleted = false;
+
+ public ValueSumPayload(Option<GenericRecord> record) {
+ super(record.isPresent() ?
Option.of(fillTotalValueIfNotPresent(record.get())) : Option.empty());
+ }
+
+ public ValueSumPayload(GenericRecord record, Comparable orderingVal) {
+ super(fillTotalValueIfNotPresent(record), orderingVal);
+ }
+
+ @Override
Review Comment:
🤖 The class name and the `sum_long_value` field suggest this payload should
accumulate a running total across updates, but `preCombine` and
`combineAndGetUpdateValue` only pick by ordering value and return the incoming
record as-is (same as the default payload) — no summation happens. Is the sum
meant to be computed here? Also `oldRecord` on line 54 is decoded but never
used.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-integration-test/src/test/java/org/apache/hudi/integration/test/TestValueSumPayload.java:
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.hudi.integration.test;
+
+import org.junit.jupiter.api.Test;
+
+public class TestValueSumPayload {
+ @Test
+ void testPrecombine() {
Review Comment:
🤖 nit: `testPrecombine()` is an empty test body — it passes vacuously and
gives no signal. Could you either add assertions or mark it with `@Disabled`
with a TODO so it's clear this is intentionally a placeholder?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-integration-test/src/main/java/org/apache/hudi/common/model/ValueSumPayload.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.hudi.common.model;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.IOException;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class ValueSumPayload extends DefaultHoodieRecordPayload {
+ private static final String INDIVIDUAL_VALUE_FIELD = "long_value";
+ private static final String TOTAL_VALUE_FIELD = "sum_long_value";
+ private AtomicBoolean isDeleteComputed = new AtomicBoolean(false);
+ private boolean isDefaultRecordPayloadDeleted = false;
+
+ public ValueSumPayload(Option<GenericRecord> record) {
+ super(record.isPresent() ?
Option.of(fillTotalValueIfNotPresent(record.get())) : Option.empty());
+ }
+
+ public ValueSumPayload(GenericRecord record, Comparable orderingVal) {
+ super(fillTotalValueIfNotPresent(record), orderingVal);
+ }
+
+ @Override
+ public OverwriteWithLatestAvroPayload
preCombine(OverwriteWithLatestAvroPayload oldValue,
+ Schema schema,
+ Properties properties) {
+ if (oldValue.recordBytes.length == 0) {
+ // use natural order for delete record
+ return this;
+ }
+ try {
+ GenericRecord oldRecord =
HoodieAvroUtils.bytesToAvro(oldValue.recordBytes, schema);
+ if (oldValue.orderingVal.compareTo(orderingVal) > 0) {
+ // pick the payload with greatest ordering value
+ return oldValue;
+ } else {
+ return this;
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord
currentValue, Schema schema, Properties properties) throws IOException {
+ if (recordBytes.length == 0) {
+ return Option.empty();
Review Comment:
🤖 nit: the commented-out line `//eventTime = updateEventTime(incomingRecord,
properties);` looks like leftover debug/WIP code — could you remove it before
merging?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-integration-test/src/main/scala/org/apache/hudi/integration/test/BatchRowSourceDataValidator.scala:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.hudi.integration.test
+
+import org.apache.hudi.common.model.HoodieCommitMetadata
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.StringUtils
+import org.apache.hudi.common.util.collection.Pair
+import org.apache.hudi.exception.HoodieIOException
+import org.apache.hudi.hadoop.fs.HadoopFSUtils.getStorageConf
+import org.apache.hudi.utilities.streamer.HoodieStreamer.CHECKPOINT_KEY
+
+import org.apache.spark.sql.SparkSession
+import org.slf4j.{Logger, LoggerFactory}
+
+import java.io.IOException
+import java.util.stream
+
+class BatchRowSourceDataValidator(spark: SparkSession,
+ sourcePathStr: String,
+ hudiTablePathStr: String) extends
Serializable {
+ val LOG: Logger =
LoggerFactory.getLogger(classOf[BatchRowSourceDataValidator]);
+
+ def validate(): Unit = {
Review Comment:
🤖 nit: `LOG` is declared `val` (effectively public) — could you make it
`private val` to match the convention in the rest of the codebase?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-integration-test/src/main/scala/org/apache/hudi/integration/test/BatchRowSourceDataValidator.scala:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.hudi.integration.test
+
+import org.apache.hudi.common.model.HoodieCommitMetadata
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.StringUtils
+import org.apache.hudi.common.util.collection.Pair
+import org.apache.hudi.exception.HoodieIOException
+import org.apache.hudi.hadoop.fs.HadoopFSUtils.getStorageConf
+import org.apache.hudi.utilities.streamer.HoodieStreamer.CHECKPOINT_KEY
+
+import org.apache.spark.sql.SparkSession
+import org.slf4j.{Logger, LoggerFactory}
+
+import java.io.IOException
+import java.util.stream
+
+class BatchRowSourceDataValidator(spark: SparkSession,
+ sourcePathStr: String,
+ hudiTablePathStr: String) extends
Serializable {
+ val LOG: Logger =
LoggerFactory.getLogger(classOf[BatchRowSourceDataValidator]);
+
+ def validate(): Unit = {
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(getStorageConf(spark.sparkContext.hadoopConfiguration))
+ .setBasePath(hudiTablePathStr)
+ .build()
+ val timestampCheckpointOpt =
getLatestCommitWithCheckpoint(metaClient.getCommitsTimeline)
+ if (timestampCheckpointOpt.isDefined) {
+ val latestCheckpoint = timestampCheckpointOpt.get.getRight
+ val actualDf = spark.read.format("hudi").load(hudiTablePathStr)
+ //actualDf.persist(StorageLevel.MEMORY_AND_DISK)
+ LOG.info("Snapshot query result schema:")
+ actualDf.printSchema()
+ LOG.info("Count of rows: " + actualDf.count())
+ LOG.info("Latest commit with checkpoint: " +
timestampCheckpointOpt.get.getLeft)
+ LOG.info("Latest checkpoint: " + latestCheckpoint)
+
+ // actualDf.unpersist()
Review Comment:
🤖 nit: there are two commented-out lines (`//actualDf.persist(...)` and `//
actualDf.unpersist()`) that look like leftover exploration — could you remove
them before merging?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
