rahil-c commented on code in PR #19698:
URL: https://github.com/apache/hudi/pull/19698#discussion_r3839460516
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java:
##########
@@ -212,7 +216,26 @@ private static Option<CloudObjectMetadata> processRow(Row
row, String storageUrl
} else {
throw new HoodieIOException("unexpected object size's type in Cloud
storage events: " + obj.getClass());
}
- return Option.of(new CloudObjectMetadata(url, size));
+ long modificationTime = row.size() > 3
+ ? epochMillis(row.get(3)) :
CloudObjectMetadata.UNKNOWN_MODIFICATION_TIME;
+ return Option.of(new CloudObjectMetadata(url, size, modificationTime));
+ }
+
+ /**
+ * Epoch millis for a notification timestamp, which both S3 and GCS report
as an ISO-8601 string.
+ * Anything unparseable yields {@link
CloudObjectMetadata#UNKNOWN_MODIFICATION_TIME} rather than
+ * failing the batch, since the value only orders writes to the same object.
+ */
+ private static long epochMillis(Object rawValue) {
+ if (rawValue == null) {
+ return CloudObjectMetadata.UNKNOWN_MODIFICATION_TIME;
+ }
+ try {
+ return Instant.parse(rawValue.toString()).toEpochMilli();
+ } catch (DateTimeParseException e) {
+ log.warn("Ignoring unparseable cloud notification timestamp {}",
rawValue);
Review Comment:
`processRow` runs inside the `mapPartitions` function, so this warns once
per row on the executors. A batch whose timestamps are all in an unexpected
format would log a line per object.
The reason I would care beyond the noise: every UNKNOWN sends `buildRow(fs,
object)` down the `buildRow(fs, path)` fallback, which stats the object. So a
format mismatch silently converts the notification path back into one
`getFileStatus` per object, which is most of the 12.3s versus 121.8s advantage
you measured. Worth logging once per batch and counting, so the degradation is
visible?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java:
##########
@@ -362,6 +416,33 @@ public static List<CloudObjectMetadata> getObjectMetadata(
.collectAsList();
}
+ /**
+ * One row per cloud object, carrying the notification timestamp when the
events have one.
+ *
+ * <p>An object written more than once inside a single batch produces one
event per write. Keying
+ * only on bucket and object key, and keeping the newest event, reads such
an object once instead
+ * of once per write. Where the metadata table predates the timestamp column
the events cannot be
+ * ordered, so this falls back to the previous behaviour of de-duplicating
on size as well.
+ */
+ private static Dataset<Row> selectDistinctObjects(Dataset<Row> events,
String bucketCol, String keyCol,
+ String sizeCol, String
timeCol) {
+ if (!Arrays.asList(events.schema().fieldNames()).contains(timeCol)) {
+ log.warn("Cloud events carry no {} column; objects rewritten within a
batch will be read once per write", timeCol);
+ return events.select(bucketCol, keyCol, sizeCol).distinct();
+ }
+ String rank = "_hoodie_event_rank";
+ // rank before projecting: the columns are nested, so selecting them first
renames them to
+ // their leaf names and the window would no longer resolve bucketCol or
keyCol.
+ // Order on the parsed instant, not the raw string: '.' sorts before 'Z',
so a lexicographic
+ // desc puts 10:00:00Z ahead of the later 10:00:00.500Z whenever precision
varies in a second.
+ return events
+ .withColumn(rank, functions.row_number().over(
+ Window.partitionBy(functions.col(bucketCol), functions.col(keyCol))
+
.orderBy(functions.to_timestamp(functions.col(timeCol)).desc_nulls_last())))
Review Comment:
Fixed in aa8f9a0, flagging it here for the record since it is a correctness
change rather than a refactor. Ranking on the raw string made the order
lexicographic: a same-second pair at `10:00:00Z` and `10:00:00.500Z` picked the
earlier write, because `'.'` sorts before `'Z'`. Because `modification_time` is
the precombine field, the true newer event then loses as stale on a later batch
and the table keeps the superseded content.
One loose end I deliberately left out of that commit: `to_timestamp` yields
null on anything it cannot parse, and those rows tie under `desc_nulls_last`,
so `row_number` still picks arbitrarily among them. A tiebreak on the key or
size would close that if you think it is worth it.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudDataFetcher.java:
##########
@@ -144,15 +152,9 @@ private Option<Dataset<Row>>
getCloudObjectDataDF(List<CloudObjectMetadata> clou
for (CloudObjectMetadata o : cloudObjectMetadata) {
totalSize += o.getSize();
Review Comment:
nit: `totalSize` is summed here for the metric and again inside
`ColumnarFileMaterializer.partitionCount`. Also, for the unstructured
materializer this metric now reports bytes *referenced*, which that class's own
javadoc argues is a poor proxy for the batch's cost, so the number means
something different depending on which materializer is in play.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java:
##########
@@ -212,7 +216,26 @@ private static Option<CloudObjectMetadata> processRow(Row
row, String storageUrl
} else {
throw new HoodieIOException("unexpected object size's type in Cloud
storage events: " + obj.getClass());
}
- return Option.of(new CloudObjectMetadata(url, size));
+ long modificationTime = row.size() > 3
Review Comment:
Concrete way this reads UNKNOWN for a whole batch: if `eventTime` is stored
as a Spark `TimestampType` rather than a string, `row.get(3).toString()` is
`2026-08-12 10:00:00.5`, which `Instant.parse` rejects. Also worth noting that
Spark's `to_timestamp` in the ranking window and `Instant.parse` here are two
different parsers, so a value Spark accepts can still land as UNKNOWN.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/UnstructuredFileMaterializer.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.utilities.sources.helpers;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.utilities.schema.SchemaProvider;
+import
org.apache.hudi.utilities.sources.helpers.unstructured.UnstructuredFileRecordBuilder;
+import
org.apache.hudi.utilities.sources.helpers.unstructured.UnstructuredFileRows;
+
+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 java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.hudi.common.util.ConfigUtils.getIntWithAltKeys;
+import static org.apache.hudi.common.util.ConfigUtils.getLongWithAltKeys;
+import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.FILE_EXTENSIONS;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.FILE_EXTENSIONS_IGNORE;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.LISTING_PARALLELISM;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.PARSE_MAX_BYTES;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.WORK_BYTES_PER_PARTITION;
+
+/**
+ * Reads cloud objects as unstructured files: each object becomes one row
carrying a BLOB column
+ * plus extracted text, metadata and chunks, the same shape the
directory-listing source produces.
+ *
+ * <p>Differs from a columnar read in all three of its responsibilities.
Objects are selected by
+ * document extension rather than by data-file format. Partitions are sized by
the bytes that will
+ * actually be parsed, because an object above {@code parse.max.bytes} is
referenced without being
+ * read and so costs almost nothing. And rows are built directly rather than
through a Spark
+ * datasource.
+ */
+public class UnstructuredFileMaterializer implements CloudObjectMaterializer {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Set<String> allowedExtensions;
+ private final Set<String> ignoredExtensions;
+ private final long parseMaxBytes;
+ private final long workBytesPerPartition;
+ private final int configuredParallelism;
+ private final int defaultParallelism;
+ private final UnstructuredFileRecordBuilder recordBuilder;
+
+ public UnstructuredFileMaterializer(TypedProperties props, JavaSparkContext
jsc) {
+ this.allowedExtensions =
UnstructuredFileRows.parseExtensions(getStringWithAltKeys(props,
FILE_EXTENSIONS, true));
+ this.ignoredExtensions =
UnstructuredFileRows.parseExtensions(getStringWithAltKeys(props,
FILE_EXTENSIONS_IGNORE, true));
+ this.parseMaxBytes = getLongWithAltKeys(props, PARSE_MAX_BYTES);
+ this.workBytesPerPartition = getLongWithAltKeys(props,
WORK_BYTES_PER_PARTITION);
+ this.configuredParallelism = getIntWithAltKeys(props, LISTING_PARALLELISM);
+ this.defaultParallelism = jsc.defaultParallelism();
+ this.recordBuilder = new UnstructuredFileRecordBuilder(props);
+ }
+
+ @Override
+ public String objectKeyPredicate(String objectKey, TypedProperties props) {
+ if (!allowedExtensions.isEmpty()) {
+ return CloudObjectsSelectorCommon.extensionPredicate(objectKey,
String.join(",", allowedExtensions));
+ }
+ if (ignoredExtensions.isEmpty()) {
+ return "";
+ }
+ // no allowlist, so select everything except the denied extensions
+ List<String> denials = new ArrayList<>();
+ for (String extension : ignoredExtensions) {
+ denials.add(String.format("%s not like '%%%s'", objectKey, extension));
+ }
+ return " and " + String.join(" and ", denials);
+ }
+
+ @Override
+ public int partitionCount(List<CloudObjectMetadata> objects, long
bytesPerPartition, int minPartitions) {
Review Comment:
This is the interesting half of the class and codecov has it at 0%. The
parseable-bytes sum, the `configuredParallelism`/`defaultParallelism` floor and
the `minPartitions` max are all worth pinning, especially since the whole point
is that it deliberately differs from the columnar sizing. Could you add a unit
test for it, and one for the denylist branch of `objectKeyPredicate`?
--
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]
