feat: add experimental remote HDFS support for native DataFusion reader by comphead · Pull Request #1359 · apache/datafusion-comet

comphead · 2025-01-31T01:00:08Z

Which issue does this PR close?

Closes #1337.
Depends on #1368

Rationale for this change

What changes are included in this PR?

How are these changes tested?

Manually starting a remote hdfs cluster and running

  test("get_struct_field with DataFusion ParquetExec - simple case - remote HDFS") {

    Seq("parquet").foreach { v1List =>
      withSQLConf(
        SQLConf.USE_V1_SOURCE_LIST.key -> v1List,
        CometConf.COMET_ENABLED.key -> "true",
        CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "false",
        CometConf.COMET_NATIVE_SCAN_IMPL.key -> CometConf.SCAN_NATIVE_DATAFUSION,
        "spark.hadoop.fs.defaultFS" -> "hdfs://namenode:9000",
        "spark.hadoop.dfs.client.use.datanode.hostname" -> "true",
        "dfs.client.use.datanode.hostname" -> "true") {

        val df = spark.read
          .parquet("hdfs://namenode:9000/user/test4")
          .select("id", "first_name", "personal_info")
        df.printSchema()
        df.explain("formatted")
        df.show(false)
        // checkSparkAnswerAndOperator(df.select("nested1.id"))
      }
    }
  }

comphead · 2025-01-31T01:01:10Z

native/core/Cargo.toml

 object_store = { workspace = true }
 url = { workspace = true }
 chrono = { workspace = true }
+datafusion-objectstore-hdfs = { git = "https://github.com/comphead/datafusion-objectstore-hdfs", branch = "master", optional = true }


@andygrove I'm keeping the updated HDFS object storage in personal repo for now, let me know if there any concerns

Is there an expected timeline for when we can move to an official release? Meantime, since we have pointed to a personal repo in the past, it is reasonable to do so for this as well (especially since this is already behind some configuration flags).

Will be addressed in #1368

comphead · 2025-01-31T01:02:06Z

native/core/src/execution/planner.rs

                assert_eq!(file_groups.len(), partition_count);

-                let object_store_url = ObjectStoreUrl::local_filesystem();
+                let object_store_url = ObjectStoreUrl::parse("hdfs://namenode:9000").unwrap();


this will be addressed in #1360

The url should be available as part of the file path passed in. (see line 1178 above)

Thanks @parthchandra it is already fixed.

comphead · 2025-01-31T01:02:56Z

native/core/src/parquet/parquet_support.rs

+    session_context: Arc<SessionContext>,
+) -> Result<(), ExecutionError> {
+    // TODO: read the namenode configuration from file schema or from spark.defaultFS
+    let url = Url::try_from("hdfs://namenode:9000").unwrap();


this will be addressed in #1360

comphead · 2025-01-31T01:04:15Z

native/core/src/parquet/parquet_support.rs

    }
 }

+#[cfg(not(feature = "hdfs"))]


hdfs cargo feature makes a conditional compilation if hdfs needed

parthchandra · 2025-01-31T01:32:36Z

native/core/src/parquet/parquet_support.rs

+pub(crate) fn register_object_store(
+    session_context: Arc<SessionContext>,
+) -> Result<(), ExecutionError> {
+    let object_store = object_store::local::LocalFileSystem::new();


It doesn't have to be only a local file system.

It depends on the feature enabled for the Comet. LocalFileSystem is by default if no specific features selected.
the annotation on this method is

#[cfg(not(feature = "hdfs"))]

This allows to plugin other features like S3, etc

This particular method is responsible for no remote feature selected e.g. for local filesystem.
If a feature selected the conditional compilation will register an object store related to the feature, like HDFS or S3

parthchandra · 2025-01-31T01:32:49Z

native/core/src/execution/planner.rs

                assert_eq!(file_groups.len(), partition_count);

-                let object_store_url = ObjectStoreUrl::local_filesystem();
+                let object_store_url = ObjectStoreUrl::parse("hdfs://namenode:9000").unwrap();


The url should be available as part of the file path passed in. (see line 1178 above)

codecov-commenter · 2025-01-31T02:46:44Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 39.32%. Comparing base (f09f8af) to head (66b6e7f).
Report is 313 commits behind head on main.

Additional details and impacted files

@@              Coverage Diff              @@
##               main    #1359       +/-   ##
=============================================
- Coverage     56.12%   39.32%   -16.81%     
- Complexity      976     2081     +1105     
=============================================
  Files           119      265      +146     
  Lines         11743    61132    +49389     
  Branches       2251    12962    +10711     
=============================================
+ Hits           6591    24040    +17449     
- Misses         4012    32587    +28575     
- Partials       1140     4505     +3365

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

comphead · 2025-02-02T18:41:36Z

@andygrove @parthchandra @mbutrovich @kazuyukitanimura can I have a review please?

parthchandra

Is there a way to 'mock' hdfs and write a unit test? I suppose using the hdfs support to read a local file should do just as well.

parthchandra · 2025-02-03T17:13:51Z

Makefile

 	./mvnw install -Prelease -DskipTests $(PROFILES)
 release:
-	cd native && RUSTFLAGS="-Ctarget-cpu=native" cargo build --release
+	cd native && RUSTFLAGS="$(RUSTFLAGS) -Ctarget-cpu=native" && RUSTFLAGS=$$RUSTFLAGS cargo build --release $(FEATURES_ARG)


$$RUSTFLAGS ? Never seen this pattern before. What does this do?

Makefile syntax lightly different from bash from what I learned.
so in order to access environment variable created on fly it is needed to access it with $$.

It gets created on fly to concatenate release specific RUSTFLAGS with ones that user can set

Thanks. Learnt something new today :)

parthchandra · 2025-02-03T17:22:19Z

native/core/src/execution/planner.rs

-                    .register_object_store(&url, Arc::new(object_store));
+                // By default, local FS object store registered
+                // if `hdfs` feature enabled then HDFS file object store registered
+                let object_store_url = register_object_store(Arc::clone(&self.session_ctx))?;


Should we update this function (get_file_path) as well?
It's currently used by NATIVE_ICEBERG_COMPAT but the goal is to unify it with COMET_DATAFUSION.

Thats a good point, to verify it we probably need to read Iceberg from HDFS which can be done in #1367

We don't need to wait for actual iceberg integration. CometScan will use COMPAT_ICEBERG if the configuration is set (That's how we are able to run the unit tests).

@comphead we can log a follow up issue to update get_file_path if you like.

Thanks @parthchandra lets create a followup ticket. Appreciate if you do it as I'm afraid I can miss some Iceberg details in ticket description

#1407. There's no detail in the PR. Can you assign to me if possible, and I'll remember to take care of it.

comphead · 2025-02-03T18:08:42Z

Thanks @parthchandra the integration test will be addressed as part of #1367. We also need to think should it be a separate flow in CI?

Makefile

wForget · 2025-02-06T02:15:54Z

native/core/src/parquet/parquet_support.rs

+pub(crate) fn register_object_store(
+    session_context: Arc<SessionContext>,
+) -> Result<ObjectStoreUrl, ExecutionError> {
+    // TODO: read the namenode configuration from file schema or from spark.defaultFS


Do we need to register object store from native_scan.file_partitions?

Thanks @wForget I'm not sure I'm getting it, do you mean the better place to register the object store will be inside file_partitions iterator loop ?

do you mean the better place to register the object store will be inside file_partitions iterator loop ?

Yes, is it possible that native scan paths correspond to multiple object stores or are different from spark.defaultFs?

for HDFS/S3 the default fs can be taken from spark.hadoop.fs.defaultFS parameter.
To support multiple object stores that is interesting idea however I'm not sure when it can be addressed

for HDFS/S3 the default fs can be taken from spark.hadoop.fs.defaultFS parameter.

Sometimes I also access other hdfs ns like:

select * from `parquet`.`hdfs://other-ns:8020/warehouse/db/table`

that is interesting scenario, I'll add a separate test case for this

kazuyukitanimura · 2025-02-15T00:36:29Z

oops, @comphead would you mind merging the latest main into this PR branch in order to resolve the conflict?

# Conflicts: # native/core/src/parquet/parquet_support.rs

comphead · 2025-02-15T20:52:05Z

oops, @comphead would you mind merging the latest main into this PR branch in order to resolve the conflict?

Resolved, please have another look

kazuyukitanimura

LGTM unless @parthchandra has more comments

kazuyukitanimura · 2025-02-17T22:34:30Z

docs/source/user-guide/datasources.md

+Provide the JRE linker path in `RUSTFLAGS`, the path can vary depending on the system. Typically JRE linker is a part of installed JDK
+
+```shell
+export JAVA_HOME="/opt/homebrew/opt/openjdk@11"


nit: is JAVA_HOME still the requirement?

parthchandra

lgtm

comphead · 2025-02-18T01:25:07Z

Thanks everyone

…er (apache#1359) * feat: add experimental remote HDFS support for native DataFusion reader

comphead commented Jan 31, 2025

View reviewed changes

parthchandra reviewed Jan 31, 2025

View reviewed changes

parthchandra reviewed Feb 3, 2025

View reviewed changes

kazuyukitanimura reviewed Feb 4, 2025

View reviewed changes

Makefile Outdated Show resolved Hide resolved

comphead marked this pull request as draft February 5, 2025 00:24

wForget reviewed Feb 6, 2025

View reviewed changes

comphead added 6 commits February 13, 2025 16:17

feat: add experimental remote HDFS support for native DataFusion reader

f099ff8

fmt

36600d2

fmt

19c8bed

fmt

d9b33aa

fmt

ad030ca

update references

cb53370

comphead force-pushed the main branch from a2130e8 to cb53370 Compare February 14, 2025 01:09

comphead marked this pull request as ready for review February 14, 2025 21:59

parthchandra mentioned this pull request Feb 15, 2025

Add hdfs experimental support for native iceberg_compat reader #1407

Open

comphead added 2 commits February 15, 2025 11:57

Merge remote-tracking branch 'upstream/main'

9b9d1e1

# Conflicts: # native/core/src/parquet/parquet_support.rs

merge

66b6e7f

comphead requested a review from kazuyukitanimura February 16, 2025 18:56

kazuyukitanimura approved these changes Feb 17, 2025

View reviewed changes

parthchandra approved these changes Feb 18, 2025

View reviewed changes

comphead merged commit de0be4b into apache:main Feb 18, 2025
74 checks passed

coderfender pushed a commit to coderfender/datafusion-comet that referenced this pull request Dec 13, 2025

feat: add experimental remote HDFS support for native DataFusion read…

fd542d6

…er (apache#1359) * feat: add experimental remote HDFS support for native DataFusion reader

Conversation

comphead commented Jan 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

How are these changes tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

comphead Jan 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov-commenter commented Jan 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

comphead commented Feb 2, 2025

Uh oh!

parthchandra left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

comphead Feb 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

comphead commented Feb 3, 2025

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kazuyukitanimura commented Feb 15, 2025

Uh oh!

comphead commented Jan 31, 2025 •

edited

Loading

comphead Jan 31, 2025 •

edited

Loading

codecov-commenter commented Jan 31, 2025 •

edited

Loading

comphead Feb 3, 2025 •

edited

Loading