Repository navigation
fix: read fs.comet.libhdfs.schemes for the scan rule from the Hadoop conf - #6838
Queued
dwsmith1983 wants to merge 21 commits into
Queued
dwsmith1983 wants to merge 21 commits into
dwsmith1983 wants to merge 21 commits into
Conversation
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Closes #6837.
Rationale for this change
CometScanRulereadfs.comet.libhdfs.schemesfrom the SQLConf entryspark.hadoop.fs.comet.libhdfs.schemes, while native routes a scheme through libhdfs from the Hadoop conf the scan forwards. Spark'sSessionState.newHadoopConfcopies SQLConf keys as they are, without dropping thespark.hadoop.prefix, so the two could disagree. A list set only incore-site.xmlmade the rule fall back on a scheme native could read, and a runtimeSET spark.hadoop.fs.comet.libhdfs.schemesmade it claim a scan native then failed to open ("Unable to recognise URL").Reading the list from the scan's conf alone is not enough either. Native picks libhdfs from the list the scan forwards, which includes reader options, but libhdfs itself builds a fresh Hadoop
Configurationfrom the executor classpath and never sees those options. A scheme added only through a reader option, with itsfs.<scheme>.implalso set there, would be claimed and then fail natively withNo FileSystem for scheme.What changes are included in this PR?
NativeConfig.resolveScanLibhdfsSchemes(sessionConf, scanConf)takes the session's Hadoop conf (newHadoopConf()) and the scan's (newHadoopConfWithOptions(relation.options), the conf the scan serializes for native). It returns the schemes listed by both, which native reads through libhdfs, and the schemes listed only by the scan. Each list defaults tohdfswhen the key is unset or blank, as nativeis_hdfs_schemedoes (NativeConfig.resolveLibhdfsSchemes).fs.comet.s3Compliant.schemeswas already read from the Hadoop conf.CometScanRuleand the Delta contrib's gate (DeltaScanSupport.declineReason) both use it, so they cannot drift:spark.hadoop.fs.comet.libhdfs.schemesdoc says it is read from the Hadoop configuration and must be set before the SparkSession is created, that a scheme listed only through a reader option is not read natively, and that libhdfs reads a listed scheme's settings, such asfs.<scheme>.impl, from the executor'score-site.xml.What users see, only where the sources disagreed (query results do not change, only native or fallback):
core-site.xmlor as the bare keyfs.comet.libhdfs.schemesnow gets the scan run natively instead of falling back.SET spark.hadoop.fs.comet.libhdfs.schemes=...no longer changes what the planner claims. A scheme it added now falls back instead of failing at execution.SETof that key that leaves outhdfsno longer makeshdfs://scans fall back, since native never saw it. Set the list at startup to change it.The list set in the Spark conf at startup reaches every place and behaves as before. Passing a scan's Hadoop options through to libhdfs is left to #6967.
#6753 adds the same
resolveLibhdfsSchemes(public there) and the same unit test, so whichever of the two lands second resolves only a small conflict.How are these changes tested?
CometScanSchemeFallbackSuitewrites a Parquet table once and, in one table-driven test, appliesCometScanRuleto the Spark plan of a read of it in five cases:fake://scan is claimed. Onmainit is declined.fake://scan falls back with the reader-options reason.spark.hadoop.key: thefake://scan falls back with "Unsupported filesystem schemes: fake". Onmainit is claimed.fake,hdfsand a reader option lists onlyhdfs: thefake://scan falls back with "Unsupported filesystem schemes: fake".fileand the same table is read throughfile:: the scan falls back with the reader-options reason, though object_store readsfilenatively.DeltaScanContribSuitewrites a Delta table once and plans a read of it in the same five cases, checking for the native Delta scan or the matching "Native Delta scan does not support" reason. Both tests run every case and fail once, naming each case that broke.NativeConfigSuitecovers both resolvers: unset, blank, and a list with spaces and mixed case forresolveLibhdfsSchemes, and the libhdfs and scan-only splits forresolveScanLibhdfsSchemes.Each case is load-bearing. With the gates using only the scan's list, the reader-option and reader-only
filecases fail. With only the session's list, the reader-option, removal, and reader-onlyfilecases fail. With the scan-only check dropped, the reader-option and reader-onlyfilecases fail. These hold in both suites.CometScanSchemeFallbackSuite, NativeConfigSuite and DeltaScanContribSuite pass on Spark 3.5 and 4.1, and the build compiles on the Spark 3.4 profile and with strict warnings on 3.5.