This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new b6c4677206b4 feat: Add source path field in table created from S3/GCS
source (#14257)
b6c4677206b4 is described below
commit b6c4677206b4ba241c4b9b85110da60fcec7e1f3
Author: Roushan Kumar <[email protected]>
AuthorDate: Mon Aug 17 13:47:24 2026 +0530
feat: Add source path field in table created from S3/GCS source (#14257)
* feat: Add source path field in table created from S3/GCS source
* address comments - move config to cloud source config
* fix config
* Use Slf4j log field instead of LOG in CloudObjectsSelectorCommon
* Add missing java.io.File import in TestCloudObjectsSelectorCommon
* review(14257): address round-1 review
- make the source path column nullable: input_file_name() is non-nullable,
which
produced a required Avro field with no default and blocked enabling the
flag on
an existing table (READER_FIELD_MISSING_DEFAULT_VALUE); Spark's "unknown
file"
empty string now maps to null
- rename the column to _hoodie_cloud_source_path, matching the
_hoodie_<source>_source_* convention of the Kafka and Kinesis sources, and
expose the constant
- append via withColumn (replace-if-present) like the path-based partition
columns
instead of prepending via selectExpr behind a hard-fail collision guard;
the
guard also blocked declaring the column in a schema provider's schema,
which is
required to keep it (Source.fetchNext replaces the row-derived provider
with
the configured one, and SourceFormatAdapter rewrites rows to its schema)
- warn when a schema provider is configured and its schema lacks the column
- add sinceVersion and document the column name, value format
(percent-encoded
URI), overwrite semantics, the schema-provider caveat and the one-way door
- drop the unrelated, incomplete config-prefix refactor in CloudSourceConfig
- tests: parameterize only partitionValueAddedToRow and
loadDatasetWithSchemaAndRepartition; replace the two collision tests and
their
fixtures with one test pinning the percent-encoded value, the overwrite
and the
nullable-with-default avro field; drop the hadoop Path import and the
unrelated
visibility churn
---------
Co-authored-by: Lokesh Jain <[email protected]>
Co-authored-by: voon <[email protected]>
---
.../hudi/utilities/config/CloudSourceConfig.java | 14 +++
.../helpers/CloudObjectsSelectorCommon.java | 16 +++
.../helpers/TestCloudObjectsSelectorCommon.java | 122 ++++++++++++++++++---
3 files changed, 136 insertions(+), 16 deletions(-)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
index 46393d64fa32..e2242e6a3c87 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
@@ -191,6 +191,20 @@ public class CloudSourceConfig extends HoodieConfig {
.sinceVersion("1.0.0")
.withDocumentation("Boolean value to allow coalesce alias columns with
actual columns while reading from source");
+ public static final ConfigProperty<Boolean> INCLUDE_SOURCE_PATH_FIELD =
ConfigProperty
+ .key(STREAMER_CONFIG_PREFIX +
"source.cloud.data.include.source.path.field")
+ .defaultValue(false)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("When enabled, appends a nullable string column named
_hoodie_cloud_source_path holding the "
+ + "fully-qualified URI of the source file each record was read from,
as returned by Spark's "
+ + "input_file_name() (percent-encoded, e.g.
s3a://bucket/dir/file%20name.json). A same-named column "
+ + "already present in the dataset is overwritten. If a schema
provider is configured "
+ + "(hoodie.streamer.schemaprovider.class), declare the column in its
schema (the source schema for the "
+ + "file-based and schema-registry providers) or it is dropped before
the write, with or without a "
+ + "transformer. Once enabled on a table, disabling it fails schema
validation on the next sync unless "
+ + "hoodie.write.set.null.for.missing.columns is enabled.");
+
public static final ConfigProperty<Boolean> CLOUD_INCREMENTAL_MERGE_SCHEMA =
ConfigProperty
.key(STREAMER_CONFIG_PREFIX + "source.cloud.data.merge.schema.enable")
.defaultValue(true)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
index 8a572044e5ab..c5a9a6e7a923 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
@@ -86,6 +86,7 @@ import static
org.apache.hudi.utilities.config.S3EventsHoodieIncrSourceConfig.S3
import static
org.apache.hudi.utilities.sources.helpers.IncrSourceHelper.coalesceOrRepartition;
import static org.apache.spark.sql.functions.input_file_name;
import static org.apache.spark.sql.functions.split;
+import static org.apache.spark.sql.functions.when;
/**
* Generic helper methods to fetch from Cloud Storage during incremental fetch
from cloud storage buckets.
@@ -100,6 +101,7 @@ public class CloudObjectsSelectorCommon {
public static final String S3_BUCKET_NAME = "s3.bucket.name";
public static final String GCS_OBJECT_KEY = "name";
public static final String GCS_OBJECT_SIZE = "size";
+ public static final String CLOUD_SOURCE_PATH_COLUMN =
"_hoodie_cloud_source_path";
private static final String SPACE_DELIMTER = " ";
private static final String GCS_PREFIX = "gs://";
@@ -343,6 +345,20 @@ public class CloudObjectsSelectorCommon {
dataset = dataset.withColumn(partitionKey,
split(split(input_file_name(), partitionPathPattern).getItem(1),
StoragePath.SEPARATOR).getItem(0));
}
}
+
+ // append the source file path if configured. input_file_name() is
non-nullable, so wrap it to make the
+ // column nullable (required to add it to an existing table); the wrapper
also maps Spark's "unknown file"
+ // empty string to null. Overwrites a same-named column, matching the
partition columns above.
+ if (getBooleanWithAltKeys(properties,
CloudSourceConfig.INCLUDE_SOURCE_PATH_FIELD)) {
+ if (rowSchema != null &&
!Arrays.asList(rowSchema.fieldNames()).contains(CLOUD_SOURCE_PATH_COLUMN)) {
+ // the streamer rewrites row sources to the configured schema
provider's schema before writing
+ log.warn("Column {} is not declared in the configured schema
provider's schema; it will be dropped before "
+ + "the write unless it is declared there",
CLOUD_SOURCE_PATH_COLUMN);
+ }
+ log.info("Adding column {} to dataset", CLOUD_SOURCE_PATH_COLUMN);
+ dataset = dataset.withColumn(CLOUD_SOURCE_PATH_COLUMN,
when(input_file_name().notEqual(""), input_file_name()));
+ }
+
dataset = coalesceOrRepartition(dataset, numPartitions);
return Option.of(dataset);
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
index e4d1e42cabef..f7cb663fa320 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
@@ -21,9 +21,12 @@ package org.apache.hudi.utilities.sources.helpers;
import org.apache.hudi.HoodieSchemaConversionUtils;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.testutils.HoodieSparkClientTestHarness;
+import org.apache.hudi.utilities.config.CloudSourceConfig;
import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
+import org.apache.hudi.utilities.schema.RowBasedSchemaProvider;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
@@ -36,10 +39,15 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
+import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
@@ -80,18 +88,21 @@ public class TestCloudObjectsSelectorCommon extends
HoodieSparkClientTestHarness
Assertions.assertEquals(Collections.singletonList(expected),
result.get().collectAsList());
}
- @Test
- public void partitionValueAddedToRow() {
- List<CloudObjectMetadata> input = Collections.singletonList(new
CloudObjectMetadata("src/test/resources/data/partitioned/country=US/state=CA/data.json",
1));
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void partitionValueAddedToRow(boolean includeSourcePathField) {
+ String dataPath =
"src/test/resources/data/partitioned/country=US/state=CA/data.json";
+ List<CloudObjectMetadata> input = Collections.singletonList(new
CloudObjectMetadata(dataPath, 1));
TypedProperties properties = new TypedProperties();
properties.put("hoodie.streamer.source.cloud.data.partition.fields.from.path",
"country,state");
+ setIncludeSourcePathField(properties, includeSourcePathField);
CloudObjectsSelectorCommon cloudObjectsSelectorCommon = new
CloudObjectsSelectorCommon(properties);
Option<Dataset<Row>> result =
cloudObjectsSelectorCommon.loadAsDataset(sparkSession, input, "json",
Option.empty(), 1);
+
Assertions.assertTrue(result.isPresent());
- Assertions.assertEquals(1, result.get().count());
- Row expected = RowFactory.create("some data", "US", "CA");
- Assertions.assertEquals(Collections.singletonList(expected),
result.get().collectAsList());
+ assertRowResult(includeSourcePathField,
Collections.singletonList(dataPath), result.get(),
+ new Object[]{"some data", "US", "CA"});
}
@Test
@@ -129,27 +140,38 @@ public class TestCloudObjectsSelectorCommon extends
HoodieSparkClientTestHarness
Assertions.assertEquals(Collections.singletonList(expected),
result.get().collectAsList());
}
- @Test
- public void loadDatasetWithSchemaAndRepartition() {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void loadDatasetWithSchemaAndRepartition(boolean includeSourcePathField) {
TypedProperties props = new TypedProperties();
-
TestCloudObjectsSelectorCommon.class.getClassLoader().getResource("schema/sample_data_schema.avsc");
String schemaFilePath =
TestCloudObjectsSelectorCommon.class.getClassLoader().getResource("schema/sample_data_schema.avsc").getPath();
props.put("hoodie.streamer.schemaprovider.source.schema.file",
schemaFilePath);
props.put("hoodie.streamer.schema.provider.class.name",
FilebasedSchemaProvider.class.getName());
props.put("hoodie.streamer.source.cloud.data.partition.fields.from.path",
"country,state");
// Setting this config so that dataset repartition happens inside
`loadAsDataset`
props.put("hoodie.streamer.source.cloud.data.partition.max.size", "1");
+ setIncludeSourcePathField(props, includeSourcePathField);
+
+ String dataPath1 =
"src/test/resources/data/partitioned/country=US/state=CA/data.json";
+ String dataPath2 =
"src/test/resources/data/partitioned/country=US/state=TX/data.json";
+ String dataPath3 =
"src/test/resources/data/partitioned/country=IND/state=TS/data.json";
+
List<CloudObjectMetadata> input = Arrays.asList(
- new
CloudObjectMetadata("src/test/resources/data/partitioned/country=US/state=CA/data.json",
1000),
- new
CloudObjectMetadata("src/test/resources/data/partitioned/country=US/state=TX/data.json",
1000),
- new
CloudObjectMetadata("src/test/resources/data/partitioned/country=IND/state=TS/data.json",
1000)
- );
+ new CloudObjectMetadata(dataPath1, 1000),
+ new CloudObjectMetadata(dataPath2, 1000),
+ new CloudObjectMetadata(dataPath3, 1000));
+
CloudObjectsSelectorCommon cloudObjectsSelectorCommon = new
CloudObjectsSelectorCommon(props);
Option<Dataset<Row>> result =
cloudObjectsSelectorCommon.loadAsDataset(sparkSession, input, "json",
Option.of(new FilebasedSchemaProvider(props, jsc)), 30);
+
Assertions.assertTrue(result.isPresent());
- List<Row> expected = Arrays.asList(RowFactory.create("some data", "US",
"CA"), RowFactory.create("some data", "US", "TX"), RowFactory.create("some
data", "IND", "TS"));
- List<Row> actual = result.get().collectAsList();
- Assertions.assertEquals(new HashSet<>(expected), new HashSet<>(actual));
+ assertRowResult(
+ includeSourcePathField,
+ Arrays.asList(dataPath1, dataPath2, dataPath3),
+ result.get(),
+ new Object[]{"some data", "US", "CA"},
+ new Object[]{"some data", "US", "TX"},
+ new Object[]{"some data", "IND", "TS"});
}
@Test
@@ -275,4 +297,72 @@ public class TestCloudObjectsSelectorCommon extends
HoodieSparkClientTestHarness
Row expected = RowFactory.create("some data", null);
Assertions.assertEquals(Collections.singletonList(expected),
result.get().collectAsList());
}
+
+ @Test
+ void sourcePathColumnIsUriEncodedAndOverwritesExistingColumn(@TempDir Path
tempDir) throws IOException {
+ // file name with a space: input_file_name() returns the percent-encoded
URI, and the fixture already
+ // carries a same-named column that must be overwritten rather than
duplicated
+ Path dataFile = tempDir.resolve("we ird.json");
+ Files.write(dataFile, Collections.singletonList(
+ "{\"data\": \"some data\", \"" +
CloudObjectsSelectorCommon.CLOUD_SOURCE_PATH_COLUMN + "\":
\"existing/path\"}"));
+ TypedProperties properties = new TypedProperties();
+ setIncludeSourcePathField(properties, true);
+ CloudObjectsSelectorCommon cloudObjectsSelectorCommon = new
CloudObjectsSelectorCommon(properties);
+ List<CloudObjectMetadata> input = Collections.singletonList(new
CloudObjectMetadata(dataFile.toString(), 1));
+ Option<Dataset<Row>> result =
cloudObjectsSelectorCommon.loadAsDataset(sparkSession, input, "json",
Option.empty(), 1);
+
+ Assertions.assertTrue(result.isPresent());
+ String expectedPath = dataFile.toUri().toString();
+ Assertions.assertTrue(expectedPath.contains("%20"), expectedPath);
+ // JSON schema inference sorts the inferred fields by name, and
overwriting a column keeps its position,
+ // so the source path column stays first here instead of being appended
+
Assertions.assertEquals(Arrays.asList(CloudObjectsSelectorCommon.CLOUD_SOURCE_PATH_COLUMN,
"data"),
+ Arrays.asList(result.get().schema().fieldNames()));
+
Assertions.assertTrue(result.get().schema().apply(CloudObjectsSelectorCommon.CLOUD_SOURCE_PATH_COLUMN).nullable());
+ // the streamer derives the writer schema from the row schema; a
non-nullable field would be a required avro
+ // field without a default and could not be added to an existing table
+ HoodieSchemaField sourcePathField =
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(
+ result.get().schema(),
RowBasedSchemaProvider.HOODIE_RECORD_STRUCT_NAME,
RowBasedSchemaProvider.HOODIE_RECORD_NAMESPACE)
+ .getField(CloudObjectsSelectorCommon.CLOUD_SOURCE_PATH_COLUMN).get();
+ Assertions.assertTrue(sourcePathField.isNullable());
+ Assertions.assertTrue(sourcePathField.hasDefaultValue());
+
Assertions.assertEquals(Collections.singletonList(RowFactory.create(expectedPath,
"some data")), result.get().collectAsList());
+ }
+
+ /**
+ * Asserts that a Dataset contains expected rows; when the source path
column is enabled it is expected
+ * to be appended last, nullable, and to hold the file URI of the row's
source file.
+ */
+ private static void assertRowResult(
+ boolean includeSourcePathField,
+ List<String> dataPaths,
+ Dataset<Row> actualResult,
+ Object[]... rowContents) {
+ Assertions.assertEquals(dataPaths.size(), rowContents.length, "dataPaths
and rowContents must align");
+ Assertions.assertEquals(rowContents.length, actualResult.count());
+ List<String> fieldNames =
Arrays.asList(actualResult.schema().fieldNames());
+
+ List<Row> expected = new ArrayList<>();
+ if (includeSourcePathField) {
+
Assertions.assertEquals(CloudObjectsSelectorCommon.CLOUD_SOURCE_PATH_COLUMN,
fieldNames.get(fieldNames.size() - 1));
+
Assertions.assertTrue(actualResult.schema().apply(CloudObjectsSelectorCommon.CLOUD_SOURCE_PATH_COLUMN).nullable());
+ for (int i = 0; i < dataPaths.size(); i++) {
+ List<Object> values = new ArrayList<>(Arrays.asList(rowContents[i]));
+ // input_file_name() returns the file URI, which java.nio's
Path.toUri() reproduces byte for byte
+ values.add(new
File(dataPaths.get(i)).getAbsoluteFile().toPath().toUri().toString());
+ expected.add(RowFactory.create(values.toArray()));
+ }
+ } else {
+
Assertions.assertFalse(fieldNames.contains(CloudObjectsSelectorCommon.CLOUD_SOURCE_PATH_COLUMN));
+ for (Object[] row : rowContents) {
+ expected.add(RowFactory.create(row));
+ }
+ }
+
+ Assertions.assertEquals(new HashSet<>(expected), new
HashSet<>(actualResult.collectAsList()));
+ }
+
+ private static void setIncludeSourcePathField(TypedProperties properties,
boolean include) {
+ properties.put(CloudSourceConfig.INCLUDE_SOURCE_PATH_FIELD.key(),
String.valueOf(include));
+ }
}
