You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2020/08/09 00:12:05 UTC
[GitHub] [hudi] garyli1019 opened a new pull request #1938: [HUDI-920] Support Incremental query for MOR table
garyli1019 opened a new pull request #1938:
URL: https://github.com/apache/hudi/pull/1938
## What is the purpose of the pull request
Support incremental query for MOR table
https://issues.apache.org/jira/browse/HUDI-920
## Brief change log
- Add MergeOnReadIncrementalRelation
- Add test on TestMORDatasource
## Verify this pull request
This change added tests and can be verified as follows:
- *Added unit tests in TestMORDatasource*
## Committer checklist
- [ ] Has a corresponding JIRA in PR title & commit
- [ ] Commit message is descriptive of the change
- [ ] CI is green
- [ ] Necessary doc changes done or have another open PR
- [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-757103037
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-756264377
>hmm, the test was passed in my local.
typically, if the test depends on some aspect of the data generated, then it can pass locally and fail in CI, because the random number seed used for data generation is different. Otherwise, it should pass. esp for checking counts etc. I have not seen it being flaky for other reasons.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-755953071
@garyli1019 any hopes for this PR to be completed in the next day or so? :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io edited a comment on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-752876566
# [Codecov](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=h1) Report
> Merging [#1938](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=desc) (7129974) into [master](https://codecov.io/gh/apache/hudi/commit/5ff8e88d58c2a6efa1046716e11dca0d61cdf872?el=desc) (5ff8e88) will **decrease** coverage by `40.24%`.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #1938 +/- ##
=============================================
- Coverage 50.28% 10.04% -40.25%
+ Complexity 2991 48 -2943
=============================================
Files 410 52 -358
Lines 18406 1852 -16554
Branches 1885 223 -1662
=============================================
- Hits 9256 186 -9070
+ Misses 8392 1653 -6739
+ Partials 758 13 -745
```
| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| hudicli | `?` | `?` | |
| hudiclient | `100.00% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudicommon | `?` | `?` | |
| hudiflink | `?` | `?` | |
| hudihadoopmr | `?` | `?` | |
| hudisparkdatasource | `?` | `?` | |
| hudisync | `?` | `?` | |
| huditimelineservice | `?` | `?` | |
| hudiutilities | `10.04% <ø> (-59.62%)` | `0.00 <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...va/org/apache/hudi/utilities/IdentitySplitter.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL0lkZW50aXR5U3BsaXR0ZXIuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-2.00%)` | |
| [...va/org/apache/hudi/utilities/schema/SchemaSet.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NjaGVtYS9TY2hlbWFTZXQuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-3.00%)` | |
| [...a/org/apache/hudi/utilities/sources/RowSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvUm93U291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| [.../org/apache/hudi/utilities/sources/AvroSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvQXZyb1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| [.../org/apache/hudi/utilities/sources/JsonSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvblNvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| [...rg/apache/hudi/utilities/sources/CsvDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvQ3N2REZTU291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-10.00%)` | |
| [...g/apache/hudi/utilities/sources/JsonDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvbkRGU1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| [...apache/hudi/utilities/sources/JsonKafkaSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvbkthZmthU291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-6.00%)` | |
| [...pache/hudi/utilities/sources/ParquetDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvUGFycXVldERGU1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-5.00%)` | |
| [...lities/schema/SchemaProviderWithPostProcessor.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NjaGVtYS9TY2hlbWFQcm92aWRlcldpdGhQb3N0UHJvY2Vzc29yLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| ... and [382 more](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree-more) | |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-688685118
Ready for review. cc: @vinothchandar @bhasudha
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r554291307
##########
File path: hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,218 @@
+/*
+ * 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
+
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.listStatusForAffectedPartitions
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
Review comment:
`IncrementalRelation` extends `TablesScan` and this extends `PrunedFilteredScan` so I think to keep them separate is better.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-757103037
Thanks @garyli1019 ! You are awesome.
Will take a pass, after CI is happy. and land,
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r554442195
##########
File path: hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,218 @@
+/*
+ * 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
+
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.listStatusForAffectedPartitions
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
Review comment:
ack
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 edited a comment on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 edited a comment on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-757588401
> Seems like this test is flaky. :( @garyli1019
>
> ```
> [INFO]
> [ERROR] Failures:
> [ERROR] TestMORDataSource.testCount:310 expected: <152> but was: <2>
> ```
>
> https://api.travis-ci.com/v3/job/470167079/log.txt
@vinothchandar This was the compaction consistency question I mentioned in the sync meeting. There are actually 2 new insert records suppose to be here, but we triggered inline compaction that compacted 150 records. Also possible that the compaction was triggered earlier. Might be fixed by #2428
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-755962801
Sounds good. Will do a review, morning time PST.
Could you please check why CI is unhappy
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r480292973
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,209 @@
+/*
+ * 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
+
+import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources.{BaseRelation, Filter, GreaterThanOrEqual, IsNotNull, LessThanOrEqual, PrunedFilteredScan, TableScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.recordLevelFilter.enabled", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", "false")
+ val pushDownFilter = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+ var requiredStructSchema = StructType(Seq())
+ requiredColumns.foreach(col => {
+ val field = tableStructSchema.find(_.name == col)
+ if (field.isDefined) {
+ requiredStructSchema = requiredStructSchema.add(field.get)
+ }
+ })
+ val requiredAvroSchema = AvroConversionUtils
+ .convertStructTypeToAvroSchema(requiredStructSchema, tableAvroSchema.getName, tableAvroSchema.getNamespace)
+ val hoodieTableState = HoodieMergeOnReadTableState(
+ tableStructSchema,
+ requiredStructSchema,
+ tableAvroSchema.toString,
+ requiredAvroSchema.toString,
+ fileIndex
+ )
+ val fullSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = tableStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+ val requiredSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = requiredStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+
+ // Follow the implementation of Spark internal HadoopRDD to handle the broadcast configuration.
+ FileSystem.getLocal(jobConf)
+ SparkHadoopUtil.get.addCredentials(jobConf)
+ val rdd = new HoodieMergeOnReadRDD(
+ sqlContext.sparkContext,
+ jobConf,
+ fullSchemaParquetReader,
+ requiredSchemaParquetReader,
+ hoodieTableState
+ )
+ rdd.asInstanceOf[RDD[Row]]
+ }
+
+ def buildFileIndex(): List[HoodieMergeOnReadFileSplit] = {
+ val affectedFileStatus = new ListBuffer[FileStatus]
+ for (commit <- commitsToReturn) {
+ val metadata: HoodieCommitMetadata = HoodieCommitMetadata.fromBytes(commitsTimelineToReturn.getInstantDetails(commit)
+ .get, classOf[HoodieCommitMetadata])
+ val idWithPath = metadata.getFileIdAndFullPaths(metaClient.getBasePath).toMap
+ idWithPath.foreach(p => {
+ val file = fs.getFileStatus(new Path(p._2))
+ affectedFileStatus += file
+ })
+ }
+ val fsView = new HoodieTableFileSystemView(metaClient,
+ commitsTimelineToReturn, affectedFileStatus.toArray)
+ val fileGroup = fsView.fetchAllStoredFileGroups().iterator().toList
+ val latestCommit = fsView.getLastInstant.get().getTimestamp
+ if (log.isDebugEnabled) {
+ fileGroup.foreach(f => log.debug(s"current file group id: " +
+ s"${f.getFileGroupId} and file slices ${f.getLatestFileSlice.get().toString}"))
+ }
+ val pathGlobPattern = optParams.getOrElse(
+ DataSourceReadOptions.INCR_PATH_GLOB_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)
+ val filteredFileGroup = if(!pathGlobPattern
+ .equals(DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)) {
+ val globMatcher = new GlobPattern("*" + pathGlobPattern)
+ fileGroup.filter(f => {
+ if (f.getLatestFileSlice.get().getBaseFile.isPresent) {
+ globMatcher.matches(f.getLatestFileSlice.get().getBaseFile.get.getPath)
+ } else {
+ globMatcher.matches(f.getLatestFileSlice.get().getLatestLogFile.get().getPath.toString)
+ }
+ })
+ } else {
+ fileGroup
+ }
+
+ filteredFileGroup.map(f => {
+ val baseFile = f.getLatestDataFile
Review comment:
IIUC, your approach here is to find all the file groups impacted in commit range, then pull affected records from the latest file slice? the mergeOnReadRDD will handle the merging of such an file slice, with the commit filters applied.
let me know @garyli1019 if I am understanding this correctly. it will help me review the code.
in the meantime, can we add a separate flag/option to turn on `new.incremental.relation=true` or sth, to control this and push the changes
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r554287262
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
+ Path basePath, List<HoodieInstant> commitsToCheck, HoodieTimeline timeline) throws IOException {
+ // Extract files touched by these commits.
+ // TODO This might need to be done in parallel like listStatus parallelism ?
Review comment:
Are you referring to RFC-15 that not being landed yet? The current implementation of `HoodieParquetInputFormat` is listing all files of affected partitions and then do the filtering later.
##########
File path: hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,218 @@
+/*
+ * 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
+
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.listStatusForAffectedPartitions
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
Review comment:
`IncrementalRelation` extends `TablesScan` and this extends `PrunedFilteredScan` so I think to keep them separate is better.
##########
File path: hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,218 @@
+/*
+ * 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
+
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.listStatusForAffectedPartitions
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.filterPushdown", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.recordLevelFilter.enabled", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", "false")
+ val pushDownFilter = {
Review comment:
yes, if the filterPushdown was disabled then the record level filter will not work.
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
+ Path basePath, List<HoodieInstant> commitsToCheck, HoodieTimeline timeline) throws IOException {
+ // Extract files touched by these commits.
+ // TODO This might need to be done in parallel like listStatus parallelism ?
+ HashMap<String, HashMap<String, FileStatus>> partitionToFileStatusesMap = new HashMap<>();
+ for (HoodieInstant commit: commitsToCheck) {
+ HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes(timeline.getInstantDetails(commit).get(),
+ HoodieCommitMetadata.class);
+ for (Map.Entry<String, List<HoodieWriteStat>> entry: commitMetadata.getPartitionToWriteStats().entrySet()) {
+ if (!partitionToFileStatusesMap.containsKey(entry.getKey())) {
+ partitionToFileStatusesMap.put(entry.getKey(), new HashMap<>());
+ }
+ for (HoodieWriteStat stat : entry.getValue()) {
+ String relativeFilePath = stat.getPath();
+ Path fullPath = relativeFilePath != null ? FSUtils.getPartitionPath(basePath, relativeFilePath) : null;
+ if (fullPath != null) {
+ if (partitionToFileStatusesMap.get(entry.getKey()).containsKey(fullPath.getName())) {
+ // If filesystem support Append. Update the FileStatus of log file if being appended.
Review comment:
fixed.
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
Review comment:
This can be shared by other engines later. If we move this to spark we need to switch to scala code, then move it back later when supporting other engines. We could save some effort if we leave it here?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar edited a comment on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar edited a comment on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-757531637
Seems like this test is flaky. :( @garyli1019
```
[INFO]
[ERROR] Failures:
[ERROR] TestMORDataSource.testCount:310 expected: <152> but was: <2>
```
https://api.travis-ci.com/v3/job/470167079/log.txt
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-739132975
Found a bug https://issues.apache.org/jira/browse/HUDI-1434
Need to fix the bug before making this PR work.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-756781554
Thanks, will update this PR before Friday night PST.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r484660354
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDD.scala
##########
@@ -102,6 +104,44 @@ class HoodieMergeOnReadRDD(@transient sc: SparkContext,
rows
}
+ private def logFileIterator(split: HoodieMergeOnReadFileSplit,
+ config: Configuration): Iterator[InternalRow] =
+ new Iterator[InternalRow] {
Review comment:
nice catch :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-757097457
@vinothchandar addressed comments. You are right, the inconsistency was caused by the test data generator. The CI was happy after I changed from `generateUpdates` to `generateUniqueUpdates`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-757376967
@garyli1019 looks like the spark3 build broke
```
mcist -DskipITs -Dscala-2.12 -Dspark3
```
leads to.
```
[INFO] /Users/vs/Code/hudi/hudi-spark-datasource/hudi-spark/src/main/java:-1: info: compiling
[INFO] /Users/vs/Code/hudi/hudi-spark-datasource/hudi-spark/src/main/scala:-1: info: compiling
[INFO] Compiling 20 source files to /Users/vs/Code/hudi/hudi-spark-datasource/hudi-spark/target/classes at 1610229821451
[ERROR] /Users/vs/Code/hudi/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala:143: error: object SparkHadoopUtil in package deploy cannot be accessed in package org.apache.spark.deploy
[ERROR] SparkHadoopUtil.get.addCredentials(jobConf)
[ERROR] ^
[ERROR] one error found
```
FYI. I will try to fix this
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-756781554
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar merged pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar merged pull request #1938:
URL: https://github.com/apache/hudi/pull/1938
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r484672414
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,209 @@
+/*
+ * 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
+
+import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources.{BaseRelation, Filter, GreaterThanOrEqual, IsNotNull, LessThanOrEqual, PrunedFilteredScan, TableScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.recordLevelFilter.enabled", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", "false")
+ val pushDownFilter = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+ var requiredStructSchema = StructType(Seq())
+ requiredColumns.foreach(col => {
+ val field = tableStructSchema.find(_.name == col)
+ if (field.isDefined) {
+ requiredStructSchema = requiredStructSchema.add(field.get)
+ }
+ })
+ val requiredAvroSchema = AvroConversionUtils
+ .convertStructTypeToAvroSchema(requiredStructSchema, tableAvroSchema.getName, tableAvroSchema.getNamespace)
+ val hoodieTableState = HoodieMergeOnReadTableState(
+ tableStructSchema,
+ requiredStructSchema,
+ tableAvroSchema.toString,
+ requiredAvroSchema.toString,
+ fileIndex
+ )
+ val fullSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = tableStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+ val requiredSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = requiredStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+
+ // Follow the implementation of Spark internal HadoopRDD to handle the broadcast configuration.
+ FileSystem.getLocal(jobConf)
+ SparkHadoopUtil.get.addCredentials(jobConf)
+ val rdd = new HoodieMergeOnReadRDD(
+ sqlContext.sparkContext,
+ jobConf,
+ fullSchemaParquetReader,
+ requiredSchemaParquetReader,
+ hoodieTableState
+ )
+ rdd.asInstanceOf[RDD[Row]]
+ }
+
+ def buildFileIndex(): List[HoodieMergeOnReadFileSplit] = {
+ val affectedFileStatus = new ListBuffer[FileStatus]
+ for (commit <- commitsToReturn) {
+ val metadata: HoodieCommitMetadata = HoodieCommitMetadata.fromBytes(commitsTimelineToReturn.getInstantDetails(commit)
+ .get, classOf[HoodieCommitMetadata])
+ val idWithPath = metadata.getFileIdAndFullPaths(metaClient.getBasePath).toMap
+ idWithPath.foreach(p => {
+ val file = fs.getFileStatus(new Path(p._2))
+ affectedFileStatus += file
+ })
+ }
+ val fsView = new HoodieTableFileSystemView(metaClient,
+ commitsTimelineToReturn, affectedFileStatus.toArray)
+ val fileGroup = fsView.fetchAllStoredFileGroups().iterator().toList
+ val latestCommit = fsView.getLastInstant.get().getTimestamp
+ if (log.isDebugEnabled) {
+ fileGroup.foreach(f => log.debug(s"current file group id: " +
+ s"${f.getFileGroupId} and file slices ${f.getLatestFileSlice.get().toString}"))
+ }
+ val pathGlobPattern = optParams.getOrElse(
+ DataSourceReadOptions.INCR_PATH_GLOB_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)
+ val filteredFileGroup = if(!pathGlobPattern
+ .equals(DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)) {
+ val globMatcher = new GlobPattern("*" + pathGlobPattern)
+ fileGroup.filter(f => {
+ if (f.getLatestFileSlice.get().getBaseFile.isPresent) {
+ globMatcher.matches(f.getLatestFileSlice.get().getBaseFile.get.getPath)
+ } else {
+ globMatcher.matches(f.getLatestFileSlice.get().getLatestLogFile.get().getPath.toString)
+ }
+ })
+ } else {
+ fileGroup
+ }
+
+ filteredFileGroup.map(f => {
+ val baseFile = f.getLatestDataFile
Review comment:
Just to make sure. This only possible for the base file right? The latest log file will always be on the latest file slice?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io edited a comment on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-752876566
# [Codecov](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=h1) Report
> Merging [#1938](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=desc) (452be51) into [master](https://codecov.io/gh/apache/hudi/commit/5ff8e88d58c2a6efa1046716e11dca0d61cdf872?el=desc) (5ff8e88) will **decrease** coverage by `40.24%`.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #1938 +/- ##
=============================================
- Coverage 50.28% 10.04% -40.25%
+ Complexity 2991 48 -2943
=============================================
Files 410 52 -358
Lines 18406 1852 -16554
Branches 1885 223 -1662
=============================================
- Hits 9256 186 -9070
+ Misses 8392 1653 -6739
+ Partials 758 13 -745
```
| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| hudicli | `?` | `?` | |
| hudiclient | `100.00% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudicommon | `?` | `?` | |
| hudiflink | `?` | `?` | |
| hudihadoopmr | `?` | `?` | |
| hudisparkdatasource | `?` | `?` | |
| hudisync | `?` | `?` | |
| huditimelineservice | `?` | `?` | |
| hudiutilities | `10.04% <ø> (-59.62%)` | `0.00 <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...va/org/apache/hudi/utilities/IdentitySplitter.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL0lkZW50aXR5U3BsaXR0ZXIuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-2.00%)` | |
| [...va/org/apache/hudi/utilities/schema/SchemaSet.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NjaGVtYS9TY2hlbWFTZXQuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-3.00%)` | |
| [...a/org/apache/hudi/utilities/sources/RowSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvUm93U291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| [.../org/apache/hudi/utilities/sources/AvroSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvQXZyb1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| [.../org/apache/hudi/utilities/sources/JsonSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvblNvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| [...rg/apache/hudi/utilities/sources/CsvDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvQ3N2REZTU291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-10.00%)` | |
| [...g/apache/hudi/utilities/sources/JsonDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvbkRGU1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| [...apache/hudi/utilities/sources/JsonKafkaSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvbkthZmthU291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-6.00%)` | |
| [...pache/hudi/utilities/sources/ParquetDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvUGFycXVldERGU1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-5.00%)` | |
| [...lities/schema/SchemaProviderWithPostProcessor.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NjaGVtYS9TY2hlbWFQcm92aWRlcldpdGhQb3N0UHJvY2Vzc29yLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| ... and [382 more](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree-more) | |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-670999899
@vinothchandar This PR is functionally working. This is able to replace the existing `IncrementalRelation`, but for the safe side, we can mark this as an experimental feature. Once we gain enough confidence, we can completely replace the existing `IncrementalRelation`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r480526513
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,209 @@
+/*
+ * 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
+
+import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources.{BaseRelation, Filter, GreaterThanOrEqual, IsNotNull, LessThanOrEqual, PrunedFilteredScan, TableScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
Review comment:
This is the tricky part of this PR. We need to force a filter on `_hoodie_commit_time`, like what we have https://github.com/apache/hudi/blob/master/hudi-spark/src/main/scala/org/apache/hudi/IncrementalRelation.scala#L165.
Currently, I couldn't find an elegant way to force the filter regardless of what query the user is running. Spark optimization sometimes will skip scanning the file and produce incorrect results(e.g. count()). If we add a filter in RDD, then we will get involved in an extra ser/deser step. The elegant way might be getting into the Spark planing, but at this point I have no idea about how to do it. The easiest way is to ask the user to add a `.filter()` when loading the dataset, but we definitely don't wanna do that :)
My approach here is to force Spark to always scan the Base file and always apply the filter:
- For the Base file, force the pushdown filter and avoid using the default ParquetFileFormat reader(baseFileIterator), which will not scan the file when the user uses `df.count()` and will produce incorrect results.
- For log file, no need to filter base on commit time.
So we need to make sure the pushdown filter was on in every run if we keep this approach, otherwise the result will be incorrect.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r554287262
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
+ Path basePath, List<HoodieInstant> commitsToCheck, HoodieTimeline timeline) throws IOException {
+ // Extract files touched by these commits.
+ // TODO This might need to be done in parallel like listStatus parallelism ?
Review comment:
Are you referring to RFC-15 that not being landed yet? The current implementation of `HoodieParquetInputFormat` is listing all files of affected partitions and then do the filtering later.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-757531637
Seems like this test is flaky. :(
```
[INFO]
[ERROR] Failures:
[ERROR] TestMORDataSource.testCount:310 expected: <152> but was: <2>
```
https://api.travis-ci.com/v3/job/470167079/log.txt
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r484701932
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -443,4 +444,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
Review comment:
Copied and changed this method from #1817 , will apply this change to that PR after merged.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io edited a comment on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-752876566
# [Codecov](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=h1) Report
> Merging [#1938](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=desc) (452be51) into [master](https://codecov.io/gh/apache/hudi/commit/5ff8e88d58c2a6efa1046716e11dca0d61cdf872?el=desc) (5ff8e88) will **decrease** coverage by `40.24%`.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #1938 +/- ##
=============================================
- Coverage 50.28% 10.04% -40.25%
+ Complexity 2991 48 -2943
=============================================
Files 410 52 -358
Lines 18406 1852 -16554
Branches 1885 223 -1662
=============================================
- Hits 9256 186 -9070
+ Misses 8392 1653 -6739
+ Partials 758 13 -745
```
| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| hudicli | `?` | `?` | |
| hudiclient | `?` | `?` | |
| hudicommon | `?` | `?` | |
| hudiflink | `?` | `?` | |
| hudihadoopmr | `?` | `?` | |
| hudisparkdatasource | `?` | `?` | |
| hudisync | `?` | `?` | |
| huditimelineservice | `?` | `?` | |
| hudiutilities | `10.04% <ø> (-59.62%)` | `0.00 <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...va/org/apache/hudi/utilities/IdentitySplitter.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL0lkZW50aXR5U3BsaXR0ZXIuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-2.00%)` | |
| [...va/org/apache/hudi/utilities/schema/SchemaSet.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NjaGVtYS9TY2hlbWFTZXQuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-3.00%)` | |
| [...a/org/apache/hudi/utilities/sources/RowSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvUm93U291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| [.../org/apache/hudi/utilities/sources/AvroSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvQXZyb1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| [.../org/apache/hudi/utilities/sources/JsonSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvblNvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| [...rg/apache/hudi/utilities/sources/CsvDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvQ3N2REZTU291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-10.00%)` | |
| [...g/apache/hudi/utilities/sources/JsonDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvbkRGU1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| [...apache/hudi/utilities/sources/JsonKafkaSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvbkthZmthU291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-6.00%)` | |
| [...pache/hudi/utilities/sources/ParquetDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvUGFycXVldERGU1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-5.00%)` | |
| [...lities/schema/SchemaProviderWithPostProcessor.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NjaGVtYS9TY2hlbWFQcm92aWRlcldpdGhQb3N0UHJvY2Vzc29yLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| ... and [382 more](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree-more) | |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-755955479
> @garyli1019 any hopes for this PR to be completed in the next day or so? :)
@vinothchandar The current version should work with the newer version commit file, but I haven't added the option to merge with the parquet file as we discussed in the meeting. Would you do a round of review?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io edited a comment on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-752876566
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r480284406
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/DataSourceOptions.scala
##########
@@ -49,6 +49,7 @@ object DataSourceReadOptions {
val QUERY_TYPE_SNAPSHOT_OPT_VAL = "snapshot"
val QUERY_TYPE_READ_OPTIMIZED_OPT_VAL = "read_optimized"
val QUERY_TYPE_INCREMENTAL_OPT_VAL = "incremental"
+ val QUERY_TYPE_MOR_INCREMENTAL_OPT_VAL = "mor_incremental"
Review comment:
we need to guard this with a flag. query types are fundamental to design. I prefer not to overload them
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r480288447
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,209 @@
+/*
+ * 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
+
+import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources.{BaseRelation, Filter, GreaterThanOrEqual, IsNotNull, LessThanOrEqual, PrunedFilteredScan, TableScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
Review comment:
should these be left to the user on how sparkSession/sqlContext is configured instead?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r554291761
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
Review comment:
This can be shared by other engines later. If we move this to spark we need to switch to scala code, then move it back later when supporting other engines. We could save some effort if we leave it here?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar merged pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar merged pull request #1938:
URL: https://github.com/apache/hudi/pull/1938
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-755963403
```
[INFO]
[ERROR] Failures:
[ERROR] TestMORDataSource.testCount:309 expected: <142> but was: <138>
```
seems related?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r480584379
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,209 @@
+/*
+ * 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
+
+import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources.{BaseRelation, Filter, GreaterThanOrEqual, IsNotNull, LessThanOrEqual, PrunedFilteredScan, TableScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.recordLevelFilter.enabled", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", "false")
+ val pushDownFilter = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+ var requiredStructSchema = StructType(Seq())
+ requiredColumns.foreach(col => {
+ val field = tableStructSchema.find(_.name == col)
+ if (field.isDefined) {
+ requiredStructSchema = requiredStructSchema.add(field.get)
+ }
+ })
+ val requiredAvroSchema = AvroConversionUtils
+ .convertStructTypeToAvroSchema(requiredStructSchema, tableAvroSchema.getName, tableAvroSchema.getNamespace)
+ val hoodieTableState = HoodieMergeOnReadTableState(
+ tableStructSchema,
+ requiredStructSchema,
+ tableAvroSchema.toString,
+ requiredAvroSchema.toString,
+ fileIndex
+ )
+ val fullSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = tableStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+ val requiredSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = requiredStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+
+ // Follow the implementation of Spark internal HadoopRDD to handle the broadcast configuration.
+ FileSystem.getLocal(jobConf)
+ SparkHadoopUtil.get.addCredentials(jobConf)
+ val rdd = new HoodieMergeOnReadRDD(
+ sqlContext.sparkContext,
+ jobConf,
+ fullSchemaParquetReader,
+ requiredSchemaParquetReader,
+ hoodieTableState
+ )
+ rdd.asInstanceOf[RDD[Row]]
+ }
+
+ def buildFileIndex(): List[HoodieMergeOnReadFileSplit] = {
+ val affectedFileStatus = new ListBuffer[FileStatus]
+ for (commit <- commitsToReturn) {
+ val metadata: HoodieCommitMetadata = HoodieCommitMetadata.fromBytes(commitsTimelineToReturn.getInstantDetails(commit)
+ .get, classOf[HoodieCommitMetadata])
+ val idWithPath = metadata.getFileIdAndFullPaths(metaClient.getBasePath).toMap
+ idWithPath.foreach(p => {
+ val file = fs.getFileStatus(new Path(p._2))
+ affectedFileStatus += file
+ })
+ }
+ val fsView = new HoodieTableFileSystemView(metaClient,
+ commitsTimelineToReturn, affectedFileStatus.toArray)
+ val fileGroup = fsView.fetchAllStoredFileGroups().iterator().toList
+ val latestCommit = fsView.getLastInstant.get().getTimestamp
+ if (log.isDebugEnabled) {
+ fileGroup.foreach(f => log.debug(s"current file group id: " +
+ s"${f.getFileGroupId} and file slices ${f.getLatestFileSlice.get().toString}"))
+ }
+ val pathGlobPattern = optParams.getOrElse(
+ DataSourceReadOptions.INCR_PATH_GLOB_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)
+ val filteredFileGroup = if(!pathGlobPattern
+ .equals(DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)) {
+ val globMatcher = new GlobPattern("*" + pathGlobPattern)
+ fileGroup.filter(f => {
+ if (f.getLatestFileSlice.get().getBaseFile.isPresent) {
+ globMatcher.matches(f.getLatestFileSlice.get().getBaseFile.get.getPath)
+ } else {
+ globMatcher.matches(f.getLatestFileSlice.get().getLatestLogFile.get().getPath.toString)
+ }
+ })
+ } else {
+ fileGroup
+ }
+
+ filteredFileGroup.map(f => {
+ val baseFile = f.getLatestDataFile
Review comment:
@garyli1019 one issue we need to make sure we handle is when the file group is pending compaction. when that happens, the base file is not present in the latest slice, but stitched together from the previous slice. @bhasudha is handling this in the Hive PR. worth taking a look and ensuring
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
codecov-io commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-752876566
# [Codecov](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=h1) Report
> Merging [#1938](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=desc) (1847bd3) into [master](https://codecov.io/gh/apache/hudi/commit/c5e8a024f6a2169f8d6256039de00a44fdb2f5a0?el=desc) (c5e8a02) will **decrease** coverage by `41.97%`.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #1938 +/- ##
=============================================
- Coverage 52.02% 10.04% -41.98%
+ Complexity 2952 48 -2904
=============================================
Files 395 52 -343
Lines 17652 1852 -15800
Branches 1809 223 -1586
=============================================
- Hits 9183 186 -8997
+ Misses 7710 1653 -6057
+ Partials 759 13 -746
```
| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| hudicli | `?` | `?` | |
| hudiclient | `100.00% <ø> (ø)` | `0.00 <ø> (ø)` | |
| hudicommon | `?` | `?` | |
| hudiflink | `?` | `?` | |
| hudihadoopmr | `?` | `?` | |
| hudisparkdatasource | `?` | `?` | |
| hudisync | `?` | `?` | |
| huditimelineservice | `?` | `?` | |
| hudiutilities | `10.04% <ø> (-59.62%)` | `0.00 <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/hudi/pull/1938?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...va/org/apache/hudi/utilities/IdentitySplitter.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL0lkZW50aXR5U3BsaXR0ZXIuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-2.00%)` | |
| [...va/org/apache/hudi/utilities/schema/SchemaSet.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NjaGVtYS9TY2hlbWFTZXQuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-3.00%)` | |
| [...a/org/apache/hudi/utilities/sources/RowSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvUm93U291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| [.../org/apache/hudi/utilities/sources/AvroSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvQXZyb1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| [.../org/apache/hudi/utilities/sources/JsonSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvblNvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| [...rg/apache/hudi/utilities/sources/CsvDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvQ3N2REZTU291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-10.00%)` | |
| [...g/apache/hudi/utilities/sources/JsonDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvbkRGU1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| [...apache/hudi/utilities/sources/JsonKafkaSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvSnNvbkthZmthU291cmNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-6.00%)` | |
| [...pache/hudi/utilities/sources/ParquetDFSSource.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NvdXJjZXMvUGFycXVldERGU1NvdXJjZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-5.00%)` | |
| [...lities/schema/SchemaProviderWithPostProcessor.java](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NjaGVtYS9TY2hlbWFQcm92aWRlcldpdGhQb3N0UHJvY2Vzc29yLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| ... and [367 more](https://codecov.io/gh/apache/hudi/pull/1938/diff?src=pr&el=tree-more) | |
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r553482746
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
Review comment:
a little bit more clearer doc? does this method obtain all the file status that were affected by the list of commits to check?
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
+ Path basePath, List<HoodieInstant> commitsToCheck, HoodieTimeline timeline) throws IOException {
+ // Extract files touched by these commits.
+ // TODO This might need to be done in parallel like listStatus parallelism ?
Review comment:
Can we redo this such that it can use the metadata table for obtaining the listing? You can see how this is done in HoodieParquetInputFormat.
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
+ Path basePath, List<HoodieInstant> commitsToCheck, HoodieTimeline timeline) throws IOException {
+ // Extract files touched by these commits.
+ // TODO This might need to be done in parallel like listStatus parallelism ?
+ HashMap<String, HashMap<String, FileStatus>> partitionToFileStatusesMap = new HashMap<>();
+ for (HoodieInstant commit: commitsToCheck) {
+ HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes(timeline.getInstantDetails(commit).get(),
+ HoodieCommitMetadata.class);
+ for (Map.Entry<String, List<HoodieWriteStat>> entry: commitMetadata.getPartitionToWriteStats().entrySet()) {
+ if (!partitionToFileStatusesMap.containsKey(entry.getKey())) {
+ partitionToFileStatusesMap.put(entry.getKey(), new HashMap<>());
+ }
+ for (HoodieWriteStat stat : entry.getValue()) {
+ String relativeFilePath = stat.getPath();
+ Path fullPath = relativeFilePath != null ? FSUtils.getPartitionPath(basePath, relativeFilePath) : null;
+ if (fullPath != null) {
+ if (partitionToFileStatusesMap.get(entry.getKey()).containsKey(fullPath.getName())) {
+ // If filesystem support Append. Update the FileStatus of log file if being appended.
Review comment:
you can just pick the latest such file now. it will have the latest log size using `getFileSizeInBytes()`. No need to do the addition here.
##########
File path: hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,218 @@
+/*
+ * 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
+
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.listStatusForAffectedPartitions
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
Review comment:
can we subclass `IncrementalRelation` and reuse some of the code esp lines 54-66 etc? worth it?
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
Review comment:
can we just have this in `hudi-spark` for now. thats the only module that needs to call this.
##########
File path: hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,218 @@
+/*
+ * 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
+
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.listStatusForAffectedPartitions
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.filterPushdown", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.recordLevelFilter.enabled", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", "false")
+ val pushDownFilter = {
Review comment:
this is the secret sauce to do the filtering at the record level, I guess
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-757588401
> Seems like this test is flaky. :( @garyli1019
>
> ```
> [INFO]
> [ERROR] Failures:
> [ERROR] TestMORDataSource.testCount:310 expected: <152> but was: <2>
> ```
>
> https://api.travis-ci.com/v3/job/470167079/log.txt
@vinothchandar This was the compaction consistency question I mentioned in the sync meeting. There are actually 2 new insert records suppose to be here, but we triggered inline compaction that compacted 150 records. This test was 2 in my local back in August but now it's 152. So I was wondering if the compaction logic was changed somehow. @n3nash any thoughts?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r554291421
##########
File path: hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,218 @@
+/*
+ * 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
+
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.listStatusForAffectedPartitions
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.filterPushdown", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.recordLevelFilter.enabled", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", "false")
+ val pushDownFilter = {
Review comment:
yes, if the filterPushdown was disabled then the record level filter will not work.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] codecov-io edited a comment on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-752876566
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r484659992
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/DataSourceOptions.scala
##########
@@ -49,6 +49,7 @@ object DataSourceReadOptions {
val QUERY_TYPE_SNAPSHOT_OPT_VAL = "snapshot"
val QUERY_TYPE_READ_OPTIMIZED_OPT_VAL = "read_optimized"
val QUERY_TYPE_INCREMENTAL_OPT_VAL = "incremental"
+ val QUERY_TYPE_MOR_INCREMENTAL_OPT_VAL = "mor_incremental"
Review comment:
This option was supposed to distinguish when the user wants to run an incremental query on the MOR table but on the parquet files only. This was not really necessary because users can just define the timestamp range to achieve the same goal. So we can just use the tableType. If COW, use the old way, if MOR, use the new relation. Don't need an extra option anymore imo.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r536497790
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,209 @@
+/*
+ * 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
+
+import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources.{BaseRelation, Filter, GreaterThanOrEqual, IsNotNull, LessThanOrEqual, PrunedFilteredScan, TableScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.recordLevelFilter.enabled", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", "false")
+ val pushDownFilter = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+ var requiredStructSchema = StructType(Seq())
+ requiredColumns.foreach(col => {
+ val field = tableStructSchema.find(_.name == col)
+ if (field.isDefined) {
+ requiredStructSchema = requiredStructSchema.add(field.get)
+ }
+ })
+ val requiredAvroSchema = AvroConversionUtils
+ .convertStructTypeToAvroSchema(requiredStructSchema, tableAvroSchema.getName, tableAvroSchema.getNamespace)
+ val hoodieTableState = HoodieMergeOnReadTableState(
+ tableStructSchema,
+ requiredStructSchema,
+ tableAvroSchema.toString,
+ requiredAvroSchema.toString,
+ fileIndex
+ )
+ val fullSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = tableStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+ val requiredSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = requiredStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+
+ // Follow the implementation of Spark internal HadoopRDD to handle the broadcast configuration.
+ FileSystem.getLocal(jobConf)
+ SparkHadoopUtil.get.addCredentials(jobConf)
+ val rdd = new HoodieMergeOnReadRDD(
+ sqlContext.sparkContext,
+ jobConf,
+ fullSchemaParquetReader,
+ requiredSchemaParquetReader,
+ hoodieTableState
+ )
+ rdd.asInstanceOf[RDD[Row]]
+ }
+
+ def buildFileIndex(): List[HoodieMergeOnReadFileSplit] = {
+ val affectedFileStatus = new ListBuffer[FileStatus]
+ for (commit <- commitsToReturn) {
+ val metadata: HoodieCommitMetadata = HoodieCommitMetadata.fromBytes(commitsTimelineToReturn.getInstantDetails(commit)
+ .get, classOf[HoodieCommitMetadata])
+ val idWithPath = metadata.getFileIdAndFullPaths(metaClient.getBasePath).toMap
+ idWithPath.foreach(p => {
+ val file = fs.getFileStatus(new Path(p._2))
+ affectedFileStatus += file
+ })
+ }
+ val fsView = new HoodieTableFileSystemView(metaClient,
+ commitsTimelineToReturn, affectedFileStatus.toArray)
+ val fileGroup = fsView.fetchAllStoredFileGroups().iterator().toList
+ val latestCommit = fsView.getLastInstant.get().getTimestamp
+ if (log.isDebugEnabled) {
+ fileGroup.foreach(f => log.debug(s"current file group id: " +
+ s"${f.getFileGroupId} and file slices ${f.getLatestFileSlice.get().toString}"))
+ }
+ val pathGlobPattern = optParams.getOrElse(
+ DataSourceReadOptions.INCR_PATH_GLOB_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)
+ val filteredFileGroup = if(!pathGlobPattern
+ .equals(DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)) {
+ val globMatcher = new GlobPattern("*" + pathGlobPattern)
+ fileGroup.filter(f => {
+ if (f.getLatestFileSlice.get().getBaseFile.isPresent) {
+ globMatcher.matches(f.getLatestFileSlice.get().getBaseFile.get.getPath)
+ } else {
+ globMatcher.matches(f.getLatestFileSlice.get().getLatestLogFile.get().getPath.toString)
+ }
+ })
+ } else {
+ fileGroup
+ }
+
+ filteredFileGroup.map(f => {
+ val baseFile = f.getLatestDataFile
Review comment:
@vinothchandar getting back to this PR... Now I get a little confused, so when we pull from a pending compaction commit, why should we care about the base file that was not in the range? We just read the log file right? From a user perspective, I don't expect to read commit earlier than the commit I defined as a starting point.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-757404344
#2425 will fix the build
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#issuecomment-756173953
> ```
> [INFO]
> [ERROR] Failures:
> [ERROR] TestMORDataSource.testCount:309 expected: <142> but was: <138>
> ```
>
> seems related?
hmm, the test was passed in my local. The last time when we were working on the MOR data source, we saw the same thing happened that Travis and local are producing different results.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r554442195
##########
File path: hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,218 @@
+/*
+ * 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
+
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.listStatusForAffectedPartitions
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
Review comment:
ack
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r554291505
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
##########
@@ -470,4 +471,45 @@ private static HoodieBaseFile refreshFileStatus(Configuration conf, HoodieBaseFi
}
}
+ /**
+ * List affected file status based on given commits.
+ * @param basePath
+ * @param commitsToCheck
+ * @param timeline
+ * @return HashMap<partitionPath, HashMap<fileName, FileStatus>>
+ * @throws IOException
+ */
+ public static HashMap<String, HashMap<String, FileStatus>> listStatusForAffectedPartitions(
+ Path basePath, List<HoodieInstant> commitsToCheck, HoodieTimeline timeline) throws IOException {
+ // Extract files touched by these commits.
+ // TODO This might need to be done in parallel like listStatus parallelism ?
+ HashMap<String, HashMap<String, FileStatus>> partitionToFileStatusesMap = new HashMap<>();
+ for (HoodieInstant commit: commitsToCheck) {
+ HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes(timeline.getInstantDetails(commit).get(),
+ HoodieCommitMetadata.class);
+ for (Map.Entry<String, List<HoodieWriteStat>> entry: commitMetadata.getPartitionToWriteStats().entrySet()) {
+ if (!partitionToFileStatusesMap.containsKey(entry.getKey())) {
+ partitionToFileStatusesMap.put(entry.getKey(), new HashMap<>());
+ }
+ for (HoodieWriteStat stat : entry.getValue()) {
+ String relativeFilePath = stat.getPath();
+ Path fullPath = relativeFilePath != null ? FSUtils.getPartitionPath(basePath, relativeFilePath) : null;
+ if (fullPath != null) {
+ if (partitionToFileStatusesMap.get(entry.getKey()).containsKey(fullPath.getName())) {
+ // If filesystem support Append. Update the FileStatus of log file if being appended.
Review comment:
fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r480529973
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,209 @@
+/*
+ * 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
+
+import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources.{BaseRelation, Filter, GreaterThanOrEqual, IsNotNull, LessThanOrEqual, PrunedFilteredScan, TableScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.recordLevelFilter.enabled", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", "false")
+ val pushDownFilter = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+ var requiredStructSchema = StructType(Seq())
+ requiredColumns.foreach(col => {
+ val field = tableStructSchema.find(_.name == col)
+ if (field.isDefined) {
+ requiredStructSchema = requiredStructSchema.add(field.get)
+ }
+ })
+ val requiredAvroSchema = AvroConversionUtils
+ .convertStructTypeToAvroSchema(requiredStructSchema, tableAvroSchema.getName, tableAvroSchema.getNamespace)
+ val hoodieTableState = HoodieMergeOnReadTableState(
+ tableStructSchema,
+ requiredStructSchema,
+ tableAvroSchema.toString,
+ requiredAvroSchema.toString,
+ fileIndex
+ )
+ val fullSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = tableStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+ val requiredSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = requiredStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+
+ // Follow the implementation of Spark internal HadoopRDD to handle the broadcast configuration.
+ FileSystem.getLocal(jobConf)
+ SparkHadoopUtil.get.addCredentials(jobConf)
+ val rdd = new HoodieMergeOnReadRDD(
+ sqlContext.sparkContext,
+ jobConf,
+ fullSchemaParquetReader,
+ requiredSchemaParquetReader,
+ hoodieTableState
+ )
+ rdd.asInstanceOf[RDD[Row]]
+ }
+
+ def buildFileIndex(): List[HoodieMergeOnReadFileSplit] = {
+ val affectedFileStatus = new ListBuffer[FileStatus]
+ for (commit <- commitsToReturn) {
+ val metadata: HoodieCommitMetadata = HoodieCommitMetadata.fromBytes(commitsTimelineToReturn.getInstantDetails(commit)
+ .get, classOf[HoodieCommitMetadata])
+ val idWithPath = metadata.getFileIdAndFullPaths(metaClient.getBasePath).toMap
+ idWithPath.foreach(p => {
+ val file = fs.getFileStatus(new Path(p._2))
+ affectedFileStatus += file
+ })
+ }
+ val fsView = new HoodieTableFileSystemView(metaClient,
+ commitsTimelineToReturn, affectedFileStatus.toArray)
+ val fileGroup = fsView.fetchAllStoredFileGroups().iterator().toList
+ val latestCommit = fsView.getLastInstant.get().getTimestamp
+ if (log.isDebugEnabled) {
+ fileGroup.foreach(f => log.debug(s"current file group id: " +
+ s"${f.getFileGroupId} and file slices ${f.getLatestFileSlice.get().toString}"))
+ }
+ val pathGlobPattern = optParams.getOrElse(
+ DataSourceReadOptions.INCR_PATH_GLOB_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)
+ val filteredFileGroup = if(!pathGlobPattern
+ .equals(DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)) {
+ val globMatcher = new GlobPattern("*" + pathGlobPattern)
+ fileGroup.filter(f => {
+ if (f.getLatestFileSlice.get().getBaseFile.isPresent) {
+ globMatcher.matches(f.getLatestFileSlice.get().getBaseFile.get.getPath)
+ } else {
+ globMatcher.matches(f.getLatestFileSlice.get().getLatestLogFile.get().getPath.toString)
+ }
+ })
+ } else {
+ fileGroup
+ }
+
+ filteredFileGroup.map(f => {
+ val baseFile = f.getLatestDataFile
Review comment:
correct, that's what I was trying to do here.
sure, keep only one incremental query type makes sense to me.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] garyli1019 commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r483551224
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
##########
@@ -0,0 +1,209 @@
+/*
+ * 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
+
+import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobPattern, Path}
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes
+import org.apache.log4j.LogManager
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.sources.{BaseRelation, Filter, GreaterThanOrEqual, IsNotNull, LessThanOrEqual, PrunedFilteredScan, TableScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext}
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Experimental.
+ * Relation, that implements the Hoodie incremental view for Merge On Read table.
+ *
+ */
+class MergeOnReadIncrementalRelation(val sqlContext: SQLContext,
+ val optParams: Map[String, String],
+ val userSchema: StructType,
+ val metaClient: HoodieTableMetaClient)
+ extends BaseRelation with PrunedFilteredScan {
+
+ private val log = LogManager.getLogger(classOf[MergeOnReadIncrementalRelation])
+ private val conf = sqlContext.sparkContext.hadoopConfiguration
+ private val jobConf = new JobConf(conf)
+ private val fs = FSUtils.getFs(metaClient.getBasePath, conf)
+ private val commitTimeline = metaClient.getCommitsAndCompactionTimeline.filterCompletedInstants()
+ if (commitTimeline.empty()) {
+ throw new HoodieException("No instants to incrementally pull")
+ }
+ if (!optParams.contains(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY)) {
+ throw new HoodieException(s"Specify the begin instant time to pull from using " +
+ s"option ${DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY}")
+ }
+
+ private val lastInstant = commitTimeline.lastInstant().get()
+ private val mergeType = optParams.getOrElse(
+ DataSourceReadOptions.REALTIME_MERGE_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_REALTIME_MERGE_OPT_VAL)
+
+ private val commitsTimelineToReturn = commitTimeline.findInstantsInRange(
+ optParams(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY),
+ optParams.getOrElse(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, lastInstant.getTimestamp))
+ log.debug(s"${commitsTimelineToReturn.getInstants.iterator().toList.map(f => f.toString).mkString(",")}")
+ private val commitsToReturn = commitsTimelineToReturn.getInstants.iterator().toList
+ private val schemaUtil = new TableSchemaResolver(metaClient)
+ private val tableAvroSchema = schemaUtil.getTableAvroSchema
+ private val tableStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(tableAvroSchema)
+ private val maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(jobConf)
+ private val fileIndex = buildFileIndex()
+
+ override def schema: StructType = tableStructSchema
+
+ override def needConversion: Boolean = false
+
+ override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ log.debug(s"buildScan requiredColumns = ${requiredColumns.mkString(",")}")
+ log.debug(s"buildScan filters = ${filters.mkString(",")}")
+ // config to ensure the push down filter for parquet will be applied.
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.recordLevelFilter.enabled", "true")
+ sqlContext.sparkSession.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", "false")
+ val pushDownFilter = {
+ val isNotNullFilter = IsNotNull(HoodieRecord.COMMIT_TIME_METADATA_FIELD)
+ val largerThanFilter = GreaterThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.head.getTimestamp)
+ val lessThanFilter = LessThanOrEqual(HoodieRecord.COMMIT_TIME_METADATA_FIELD, commitsToReturn.last.getTimestamp)
+ filters :+isNotNullFilter :+ largerThanFilter :+ lessThanFilter
+ }
+ var requiredStructSchema = StructType(Seq())
+ requiredColumns.foreach(col => {
+ val field = tableStructSchema.find(_.name == col)
+ if (field.isDefined) {
+ requiredStructSchema = requiredStructSchema.add(field.get)
+ }
+ })
+ val requiredAvroSchema = AvroConversionUtils
+ .convertStructTypeToAvroSchema(requiredStructSchema, tableAvroSchema.getName, tableAvroSchema.getNamespace)
+ val hoodieTableState = HoodieMergeOnReadTableState(
+ tableStructSchema,
+ requiredStructSchema,
+ tableAvroSchema.toString,
+ requiredAvroSchema.toString,
+ fileIndex
+ )
+ val fullSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = tableStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+ val requiredSchemaParquetReader = new ParquetFileFormat().buildReaderWithPartitionValues(
+ sparkSession = sqlContext.sparkSession,
+ dataSchema = tableStructSchema,
+ partitionSchema = StructType(Nil),
+ requiredSchema = requiredStructSchema,
+ filters = pushDownFilter,
+ options = optParams,
+ hadoopConf = sqlContext.sparkSession.sessionState.newHadoopConf()
+ )
+
+ // Follow the implementation of Spark internal HadoopRDD to handle the broadcast configuration.
+ FileSystem.getLocal(jobConf)
+ SparkHadoopUtil.get.addCredentials(jobConf)
+ val rdd = new HoodieMergeOnReadRDD(
+ sqlContext.sparkContext,
+ jobConf,
+ fullSchemaParquetReader,
+ requiredSchemaParquetReader,
+ hoodieTableState
+ )
+ rdd.asInstanceOf[RDD[Row]]
+ }
+
+ def buildFileIndex(): List[HoodieMergeOnReadFileSplit] = {
+ val affectedFileStatus = new ListBuffer[FileStatus]
+ for (commit <- commitsToReturn) {
+ val metadata: HoodieCommitMetadata = HoodieCommitMetadata.fromBytes(commitsTimelineToReturn.getInstantDetails(commit)
+ .get, classOf[HoodieCommitMetadata])
+ val idWithPath = metadata.getFileIdAndFullPaths(metaClient.getBasePath).toMap
+ idWithPath.foreach(p => {
+ val file = fs.getFileStatus(new Path(p._2))
+ affectedFileStatus += file
+ })
+ }
+ val fsView = new HoodieTableFileSystemView(metaClient,
+ commitsTimelineToReturn, affectedFileStatus.toArray)
+ val fileGroup = fsView.fetchAllStoredFileGroups().iterator().toList
+ val latestCommit = fsView.getLastInstant.get().getTimestamp
+ if (log.isDebugEnabled) {
+ fileGroup.foreach(f => log.debug(s"current file group id: " +
+ s"${f.getFileGroupId} and file slices ${f.getLatestFileSlice.get().toString}"))
+ }
+ val pathGlobPattern = optParams.getOrElse(
+ DataSourceReadOptions.INCR_PATH_GLOB_OPT_KEY,
+ DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)
+ val filteredFileGroup = if(!pathGlobPattern
+ .equals(DataSourceReadOptions.DEFAULT_INCR_PATH_GLOB_OPT_VAL)) {
+ val globMatcher = new GlobPattern("*" + pathGlobPattern)
+ fileGroup.filter(f => {
+ if (f.getLatestFileSlice.get().getBaseFile.isPresent) {
+ globMatcher.matches(f.getLatestFileSlice.get().getBaseFile.get.getPath)
+ } else {
+ globMatcher.matches(f.getLatestFileSlice.get().getLatestLogFile.get().getPath.toString)
+ }
+ })
+ } else {
+ fileGroup
+ }
+
+ filteredFileGroup.map(f => {
+ val baseFile = f.getLatestDataFile
Review comment:
I think we can share the implementation of this part for Hive and Spark. I will move this part to a util class.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #1938: [HUDI-920] Support Incremental query for MOR table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1938:
URL: https://github.com/apache/hudi/pull/1938#discussion_r480286220
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDD.scala
##########
@@ -102,6 +104,44 @@ class HoodieMergeOnReadRDD(@transient sc: SparkContext,
rows
}
+ private def logFileIterator(split: HoodieMergeOnReadFileSplit,
+ config: Configuration): Iterator[InternalRow] =
+ new Iterator[InternalRow] {
Review comment:
this does not seem to be indented correctly?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org