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 2021/03/25 11:34:37 UTC

[GitHub] [hudi] xiarixiaoyao opened a new pull request #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

xiarixiaoyao opened a new pull request #2721:
URL: https://github.com/apache/hudi/pull/2721


   
   ## *Tips*
   - *Thank you very much for contributing to Apache Hudi.*
   - *Please review https://hudi.apache.org/contributing.html before opening a pull request.*
   
   ## What is the purpose of the pull request
   
   Fix the StackOverflowError on [HUDI-1720] 
   
   now RealtimeCompactedRecordReader.next   deal with delete records by recursion, see:
   https://github.com/apache/hudi/blob/6e803e08b1328b32a5c3a6acd8168fdabc8a1e50/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java#L106
   however when the log file contains many delete record,  the logcial of RealtimeCompactedRecordReader.next  will lead stackOverflowError
   
   we can use Loop instead of recursion。
   
   ## Brief change log
   
   *(for example:)*
     - *Modify AnnotationLocation checkstyle rule in checkstyle.xml*
   
   ## Verify this pull request
   Manually verified the change by running a job locally
   
   ## 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] xiarixiaoyao edited a comment on pull request #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

Posted by GitBox <gi...@apache.org>.
xiarixiaoyao edited a comment on pull request #2721:
URL: https://github.com/apache/hudi/pull/2721#issuecomment-806582989


   test step:
   
   before patch：
   
   step1:
   
   val df = spark.range(0, 1000000).toDF("keyid")
   .withColumn("col3", expr("keyid + 10000000"))
   .withColumn("p", lit(0))
   .withColumn("p1", lit(0))
   .withColumn("p2", lit(7))
   .withColumn("a1", lit(Array[String] ("sb1", "rz")))
   .withColumn("a2", lit(Array[String] ("sb1", "rz")))
   
   // bulk_insert 100w row (keyid from 0 to 1000000)
   
   merge(df, 4, "default", "hive_9b", DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, op = "bulk_insert")
   
   step2:
   
   val df = spark.range(0, 900000).toDF("keyid")
   .withColumn("col3", expr("keyid + 10000000"))
   .withColumn("p", lit(0))
   .withColumn("p1", lit(0))
   .withColumn("p2", lit(7))
   .withColumn("a1", lit(Array[String] ("sb1", "rz")))
   .withColumn("a2", lit(Array[String] ("sb1", "rz")))
   
   // delete 90w row (keyid from 0 to 900000)
   
   delete(df, 4, "default", "hive_9b")
   
   step3:
   
   query on beeline/spark-sql :  select count(col3)  from hive_9b_rt
   2021-03-25 15:33:29,029 | INFO  | main | RECORDS_OUT_OPERATOR_RS_3:1, RECORDS_OUT_INTERMEDIATE:1,  | Operator.java:10382021-03-25 15:33:29,029 | INFO  | main | RECORDS_OUT_OPERATOR_RS_3:1, RECORDS_OUT_INTERMEDIATE:1,  | Operator.java:10382021-03-25 15:33:29,029 | ERROR | main | Error running child : java.lang.StackOverflowError at org.apache.parquet.bytes.BytesUtils.readIntLittleEndian(BytesUtils.java:83) at org.apache.parquet.column.values.plain.BinaryPlainValuesReader.readBytes(BinaryPlainValuesReader.java:39) at org.apache.parquet.column.impl.ColumnReaderBase$2$6.read(ColumnReaderBase.java:344) at org.apache.parquet.column.impl.ColumnReaderBase.readValue(ColumnReaderBase.java:503) at org.apache.parquet.column.impl.ColumnReaderImpl.readValue(ColumnReaderImpl.java:30) at org.apache.parquet.column.impl.ColumnReaderBase.writeCurrentValueToConverter(ColumnReaderBase.java:409) at org.apache.parquet.column.impl.ColumnReaderImpl.writeCurrentValueToConverter(ColumnReaderImpl.java:30) at
  org.apache.parquet.io.RecordReaderImplementation.read(RecordReaderImplementation.java:406) at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:226) at org.apache.parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:207) at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.next(ParquetRecordReaderWrapper.java:159) at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.next(ParquetRecordReaderWrapper.java:41) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:84) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.Real
 timeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCom
 pactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106)
   
   After patch：
   	select count(col3) from hive_9b_rt
   	+---------+
   	|   _c0   |
   	+---------+
   	| 100000  |
   	+---------+
   step4： 
       val df = spark.range(900000, 1000000).toDF("keyid")
         .withColumn("col3", expr("keyid + 10000000"))
         .withColumn("p", lit(0))
         .withColumn("p1", lit(0))
         .withColumn("p2", lit(7))
         .withColumn("a1", lit(Array[String] ("sb1", "rz")))
         .withColumn("a2", lit(Array[String] ("sb1", "rz")))
   // delete remaining 10W lines
   delete(df, 4, "default", "hive_9b")
   sparksql/hive-beeline:
   select count(col3) from hive_9b_rt;
   +------+
   | _c0  |
   +------+
   | 0    |
   +------+
   
   delete function:
     def delete(df: DataFrame, par: Int, db: String, tableName: String, tableType: String = DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL): Unit = {
       df.write.format("hudi").
         option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, tableType).
         option(HoodieCompactionConfig.INLINE_COMPACT_PROP, false).
         option(PRECOMBINE_FIELD_OPT_KEY, "col3").
         option(RECORDKEY_FIELD_OPT_KEY, "keyid").
         option(PARTITIONPATH_FIELD_OPT_KEY, "p,p1,p2").
         option(HoodieWriteConfig.KEYGENERATOR_CLASS_PROP, classOf[ComplexKeyGenerator].getName).
         option(DataSourceWriteOptions.OPERATION_OPT_KEY, "delete").
         option("hoodie.bulkinsert.shuffle.parallelism", par.toString).
         option("hoodie.insert.shuffle.parallelism", par.toString).
         option("hoodie.upsert.shuffle.parallelism", par.toString).
         option("hoodie.delete.shuffle.parallelism", par.toString).
         option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY, "true").
         option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY, "p,p1,p2").
         option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY, "org.apache.hudi.hive.MultiPartKeysValueExtractor").
         option("hoodie.datasource.hive_sync.support_timestamp", "true").
         option(HIVE_STYLE_PARTITIONING_OPT_KEY, "true").
         option(HIVE_USE_JDBC_OPT_KEY, "false").
         option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY, db).
         option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY, tableName).
         option(TABLE_NAME, tableName).mode(Append).save(s"/tmp/${db}/${tableName}")
     }
   
   merge function:
   def merge(df: org.apache.spark.sql.DataFrame, par: Int, db: String, tableName: String,
   tableType: String = DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL,
   hivePartitionExtract: String = "org.apache.hudi.hive.MultiPartKeysValueExtractor", op: String = "upsert"): Unit = {
   val mode = if (op.equals("bulk_insert")) {
   Overwrite
   } else {
   Append
   }
   df.write.format("hudi").
   option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, tableType).
   option(HoodieCompactionConfig.INLINE_COMPACT_PROP, false).
   option(PRECOMBINE_FIELD_OPT_KEY, "col3").
   option(RECORDKEY_FIELD_OPT_KEY, "keyid").
   option(PARTITIONPATH_FIELD_OPT_KEY, "p,p1,p2").
   option(DataSourceWriteOptions.OPERATION_OPT_KEY, op).
   option(HoodieWriteConfig.KEYGENERATOR_CLASS_PROP, classOf[ComplexKeyGenerator].getName).
   option("hoodie.bulkinsert.shuffle.parallelism", par.toString).
   option("hoodie.metadata.enable", "false").
   option("hoodie.insert.shuffle.parallelism", par.toString).
   option("hoodie.upsert.shuffle.parallelism", par.toString).
   option("hoodie.delete.shuffle.parallelism", par.toString).
   option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY, "true").
   option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY, "p,p1,p2").
   option("hoodie.datasource.hive_sync.support_timestamp", "true").
   option(HIVE_STYLE_PARTITIONING_OPT_KEY, "true").
   option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY, "org.apache.hudi.hive.MultiPartKeysValueExtractor").
   option(HIVE_USE_JDBC_OPT_KEY, "false").
   option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY, db).
   option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY, tableName).
   option(TABLE_NAME, tableName).mode(mode).save(s"/tmp/${db}/${tableName}")
   }
   


-- 
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] nsivabalan commented on pull request #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

Posted by GitBox <gi...@apache.org>.
nsivabalan commented on pull request #2721:
URL: https://github.com/apache/hudi/pull/2721#issuecomment-810529468


   @garyli1019 : fix the label as required. (sev:critical or sev:high). Also fix the corresponding jira if applicable. 


-- 
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] xiarixiaoyao commented on a change in pull request #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

Posted by GitBox <gi...@apache.org>.
xiarixiaoyao commented on a change in pull request #2721:
URL: https://github.com/apache/hudi/pull/2721#discussion_r601949535



##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java
##########
@@ -95,15 +103,24 @@ public boolean next(NullWritable aVoid, ArrayWritable arrayWritable) throws IOEx
         // TODO(NA): Invoke preCombine here by converting arrayWritable to Avro. This is required since the
         // deltaRecord may not be a full record and needs values of columns from the parquet
         Option<GenericRecord> rec;
-        if (usesCustomPayload) {
-          rec = deltaRecordMap.get(key).getData().getInsertValue(getWriterSchema());
-        } else {
-          rec = deltaRecordMap.get(key).getData().getInsertValue(getReaderSchema());
+        rec = buildGenericRecordwithCustomPayload(deltaRecordMap.get(key));
+        // If the record is not present, this is a delete record using an empty payload so skip this base record
+        // and move to the next record
+        while (!rec.isPresent()) {
+          // if current parquet reader has no record, return false
+          if (!this.parquetReader.next(aVoid, arrayWritable)) {

Review comment:
       @garyli1019  thanks for your replay.       no, we will not miss a record.  when we call this.parquetReader.next(aVoid, arrayWritable) and this function return true,  parquet reader wil fill the new record to arrayWritable auto(see parquet reader source code), 
   
   source code of parquet reader:
     public boolean next(final NullWritable key, final ArrayWritable value) throws IOException {
           ************************
         ****************
           if (value != null && arrValue.length == arrCurrent.length) {
             ****System.arraycopy(arrCurrent, 0, arrValue, 0, arrCurrent.length);****
           } else {
             if (arrValue.length != arrCurrent.length) {
               throw new IOException("DeprecatedParquetHiveInput : size of object differs. Value" +
                 " size :  " + arrValue.length + ", Current Object size : " + arrCurrent.length);
             } else {
               throw new IOException("DeprecatedParquetHiveInput can not support RecordReaders that" +
                 " don't return same key & value & value is null");
             }
           }




-- 
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 #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #2721:
URL: https://github.com/apache/hudi/pull/2721#discussion_r605342627



##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java
##########
@@ -95,15 +103,24 @@ public boolean next(NullWritable aVoid, ArrayWritable arrayWritable) throws IOEx
         // TODO(NA): Invoke preCombine here by converting arrayWritable to Avro. This is required since the
         // deltaRecord may not be a full record and needs values of columns from the parquet
         Option<GenericRecord> rec;
-        if (usesCustomPayload) {
-          rec = deltaRecordMap.get(key).getData().getInsertValue(getWriterSchema());
-        } else {
-          rec = deltaRecordMap.get(key).getData().getInsertValue(getReaderSchema());
+        rec = buildGenericRecordwithCustomPayload(deltaRecordMap.get(key));
+        // If the record is not present, this is a delete record using an empty payload so skip this base record
+        // and move to the next record
+        while (!rec.isPresent()) {
+          // if current parquet reader has no record, return false
+          if (!this.parquetReader.next(aVoid, arrayWritable)) {

Review comment:
       ok, I got confused by Spark Record Reader Iterator with this. There is no problem 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] xiarixiaoyao commented on pull request #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

Posted by GitBox <gi...@apache.org>.
xiarixiaoyao commented on pull request #2721:
URL: https://github.com/apache/hudi/pull/2721#issuecomment-806584382


   @garyli1019  could you help me review this pr, thanks


-- 
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 merged pull request #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

Posted by GitBox <gi...@apache.org>.
garyli1019 merged pull request #2721:
URL: https://github.com/apache/hudi/pull/2721


   


-- 
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] xiarixiaoyao commented on pull request #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

Posted by GitBox <gi...@apache.org>.
xiarixiaoyao commented on pull request #2721:
URL: https://github.com/apache/hudi/pull/2721#issuecomment-806582989


   test step:
   
   before patch：
   
   step1:
   
   val df = spark.range(0, 1000000).toDF("keyid")
   .withColumn("col3", expr("keyid + 10000000"))
   .withColumn("p", lit(0))
   .withColumn("p1", lit(0))
   .withColumn("p2", lit(7))
   .withColumn("a1", lit(Array[String] ("sb1", "rz")))
   .withColumn("a2", lit(Array[String] ("sb1", "rz")))
   
   // bulk_insert 100w row (keyid from 0 to 1000000)
   
   merge(df, 4, "default", "hive_9b", DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, op = "bulk_insert")
   
   step2:
   
   val df = spark.range(0, 900000).toDF("keyid")
   .withColumn("col3", expr("keyid + 10000000"))
   .withColumn("p", lit(0))
   .withColumn("p1", lit(0))
   .withColumn("p2", lit(7))
   .withColumn("a1", lit(Array[String] ("sb1", "rz")))
   .withColumn("a2", lit(Array[String] ("sb1", "rz")))
   
   // delete 90w row (keyid from 0 to 900000)
   
   delete(df, 4, "default", "hive_9b")
   
   step3:
   
   query on beeline/spark-sql :  select count(col3)  from hive_9b_rt
   2021-03-25 15:33:29,029 | INFO  | main | RECORDS_OUT_OPERATOR_RS_3:1, RECORDS_OUT_INTERMEDIATE:1,  | Operator.java:10382021-03-25 15:33:29,029 | INFO  | main | RECORDS_OUT_OPERATOR_RS_3:1, RECORDS_OUT_INTERMEDIATE:1,  | Operator.java:10382021-03-25 15:33:29,029 | ERROR | main | Error running child : java.lang.StackOverflowError at org.apache.parquet.bytes.BytesUtils.readIntLittleEndian(BytesUtils.java:83) at org.apache.parquet.column.values.plain.BinaryPlainValuesReader.readBytes(BinaryPlainValuesReader.java:39) at org.apache.parquet.column.impl.ColumnReaderBase$2$6.read(ColumnReaderBase.java:344) at org.apache.parquet.column.impl.ColumnReaderBase.readValue(ColumnReaderBase.java:503) at org.apache.parquet.column.impl.ColumnReaderImpl.readValue(ColumnReaderImpl.java:30) at org.apache.parquet.column.impl.ColumnReaderBase.writeCurrentValueToConverter(ColumnReaderBase.java:409) at org.apache.parquet.column.impl.ColumnReaderImpl.writeCurrentValueToConverter(ColumnReaderImpl.java:30) at
  org.apache.parquet.io.RecordReaderImplementation.read(RecordReaderImplementation.java:406) at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:226) at org.apache.parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:207) at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.next(ParquetRecordReaderWrapper.java:159) at org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper.next(ParquetRecordReaderWrapper.java:41) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:84) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.Real
 timeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCom
 pactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106) at org.apache.hudi.hadoop.realtime.RealtimeCompactedRecordReader.next(RealtimeCompactedRecordReader.java:106)
   
   After patch：
   	select count(col3) from hive_9b_rt
   	+---------+
   	|   _c0   |
   	+---------+
   	| 100000  |
   	+---------+
   step4： 
       val df = spark.range(900000, 1000000).toDF("keyid")
         .withColumn("col3", expr("keyid + 10000000"))
         .withColumn("p", lit(0))
         .withColumn("p1", lit(0))
         .withColumn("p2", lit(7))
         .withColumn("a1", lit(Array[String] ("sb1", "rz")))
         .withColumn("a2", lit(Array[String] ("sb1", "rz")))
   // delete remaining 10W lines
   delete(df, 4, "default", "hive_9b")
   sparksql/hive-beeline:
   select count(col3) from hive_9b_rt;
   +------+
   | _c0  |
   +------+
   | 0    |
   +------+


-- 
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 #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

Posted by GitBox <gi...@apache.org>.
codecov-io commented on pull request #2721:
URL: https://github.com/apache/hudi/pull/2721#issuecomment-806823048


   # [Codecov](https://codecov.io/gh/apache/hudi/pull/2721?src=pr&el=h1) Report
   > Merging [#2721](https://codecov.io/gh/apache/hudi/pull/2721?src=pr&el=desc) (fb529db) into [master](https://codecov.io/gh/apache/hudi/commit/6e803e08b1328b32a5c3a6acd8168fdabc8a1e50?el=desc) (6e803e0) will **decrease** coverage by `42.32%`.
   > The diff coverage is `n/a`.
   
   [![Impacted file tree graph](https://codecov.io/gh/apache/hudi/pull/2721/graphs/tree.svg?width=650&height=150&src=pr&token=VTTXabwbs2)](https://codecov.io/gh/apache/hudi/pull/2721?src=pr&el=tree)
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #2721       +/-   ##
   ============================================
   - Coverage     51.72%   9.40%   -42.33%     
   + Complexity     3601      48     -3553     
   ============================================
     Files           476      54      -422     
     Lines         22595    1989    -20606     
     Branches       2409     236     -2173     
   ============================================
   - Hits          11687     187    -11500     
   + Misses         9889    1789     -8100     
   + Partials       1019      13     -1006     
   ```
   
   | Flag | Coverage Δ | Complexity Δ | |
   |---|---|---|---|
   | hudicli | `?` | `?` | |
   | hudiclient | `?` | `?` | |
   | hudicommon | `?` | `?` | |
   | hudiflink | `?` | `?` | |
   | hudihadoopmr | `?` | `?` | |
   | hudisparkdatasource | `?` | `?` | |
   | hudisync | `?` | `?` | |
   | huditimelineservice | `?` | `?` | |
   | hudiutilities | `9.40% <ø> (-60.34%)` | `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/2721?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
   |---|---|---|---|
   | [...va/org/apache/hudi/utilities/IdentitySplitter.java](https://codecov.io/gh/apache/hudi/pull/2721/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/2721/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/2721/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/2721/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/2721/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/2721/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/2721/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/2721/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/2721/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/2721/diff?src=pr&el=tree#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3NjaGVtYS9TY2hlbWFQcm92aWRlcldpdGhQb3N0UHJvY2Vzc29yLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
   | ... and [452 more](https://codecov.io/gh/apache/hudi/pull/2721/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 a change in pull request #2721: [HUDI-1720] when query incr view of mor table which has many delete records use sparksql/hive-beeline, StackOverflowError

Posted by GitBox <gi...@apache.org>.
garyli1019 commented on a change in pull request #2721:
URL: https://github.com/apache/hudi/pull/2721#discussion_r601540575



##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java
##########
@@ -95,15 +103,24 @@ public boolean next(NullWritable aVoid, ArrayWritable arrayWritable) throws IOEx
         // TODO(NA): Invoke preCombine here by converting arrayWritable to Avro. This is required since the
         // deltaRecord may not be a full record and needs values of columns from the parquet
         Option<GenericRecord> rec;
-        if (usesCustomPayload) {
-          rec = deltaRecordMap.get(key).getData().getInsertValue(getWriterSchema());
-        } else {
-          rec = deltaRecordMap.get(key).getData().getInsertValue(getReaderSchema());
+        rec = buildGenericRecordwithCustomPayload(deltaRecordMap.get(key));
+        // If the record is not present, this is a delete record using an empty payload so skip this base record
+        // and move to the next record
+        while (!rec.isPresent()) {
+          // if current parquet reader has no record, return false
+          if (!this.parquetReader.next(aVoid, arrayWritable)) {

Review comment:
       if parquet has records, this will get the record but we didn't read it, so I guess we will miss a record 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