Fix comment in RFC46

[wzx140]

Change Logs

Fix comments in rfc46. You can check all commits.

Impact

Describe any public API or user-facing feature change or any performance impact.

Risk level: none | low | medium | high

Choose one. If medium or high, explain what verification was done to mitigate the risks.

Contributor's checklist

• Read through contributor's guide
• Change Logs and Impact were stated clearly
• Adequate tests were added if applicable
• CI passed

[wzx140]

@alexeykudinkin Could you please review it? Really thanks! Geng Xiaoyu and me have fix comments in RFC46. I found @DefaultSerializer(HoodieSparkRecordSerializer.class) and user need not to set spark.kryo.registrator.

[wzx140]

https://dev.azure.com/apache-hudi-ci-org/apache-hudi-ci/_build/results?buildId=11609&view=results

[alexeykudinkin]

@wzx140 can you please validate that @DefaultSerializer(HoodieSparkRecordSerializer.class) leads to HoodieSparkRecordSerializer being used by Kryo when shuffling?

[alexeykudinkin]

@wzx140 why do we still need to do newInstance why can't we use HoodieRecord returned by the merger?

[wzx140]

We can see the old logic. It make a new record that use combinedValue and other parts from hoodieRecord.

      HoodieRecord<? extends HoodieRecordPayload> oldRecord = records.get(key);
      HoodieRecordPayload oldValue = oldRecord.getData();
      HoodieRecordPayload combinedValue = hoodieRecord.getData().preCombine(oldValue);
      // If combinedValue is oldValue, no need rePut oldRecord
      if (combinedValue != oldValue) {
        HoodieOperation operation = hoodieRecord.getOperation();
        records.put(key, new HoodieAvroRecord<>(new HoodieKey(key, hoodieRecord.getPartitionPath()), combinedValue, operation));
      }

[alexeykudinkin]

But that doesn't make sense, right? Previously preCombine would return HoodieRecordPayload we're now returning HoodieRecord then call getData and immediately write it back into HoodieRecord.

Let's simplify it

[alexeykudinkin]

Let's not forget to change the interface to return Object[]

[alexeykudinkin]

You can drop ()

[alexeykudinkin]

Sorry, i skimmed t/h this initially but overlooked how we're doing the serialization here:

• We should only serialize the UnsafeRow that HoodieSparkRecord holds (see my comment above); UnsafeRow implements KryoSerializable so we can use these methods to properly write it out into the stream; we should not be relying on Java's object serialization framework for that

[alexeykudinkin]

Shall we call this ComparableLists? "Flat" reference is confusing

[wzx140]

It is copied from calcite core. And ComparableLists already exists in FlatLists.java.

[wzx140]

@wzx140 can you please validate that @DefaultSerializer(HoodieSparkRecordSerializer.class) leads to HoodieSparkRecordSerializer being used by Kryo when shuffling?

Yes. I check it in UT with printing some message in HoodieSparkRecordSerializer.

[wzx140]

@hudi-bot run azure

[wzx140]

@alexeykudinkin Please review the update when you have time.

[alexeykudinkin]

But that doesn't make sense, right? Previously preCombine would return HoodieRecordPayload we're now returning HoodieRecord then call getData and immediately write it back into HoodieRecord.

Let's simplify it

[alexeykudinkin]

Why are we doing that?

[wzx140]

Because in avro2parquetLog AvroWriteSupport#"parquet.avro.write-old-list-structure" default value is true and currently we use its default value. In order to be compatible, We should also let sparkRecord2parquetLog write with legency format

[alexeykudinkin]

Sorry, not sure what avro2parquetLog is

[wzx140]

If we use HoodieAvroRecord to write parquetLog, it will write with AvroWriteSupport#"parquet.avro.write-old-list-structure"=true. In order to be compatible, We should also use HoodieSparkRecord write parquetLog with legency format. So that we can use HoodieAvroRecord to read the parquet log which written by HoodieSparkRecord.

        // Standard mode:
        //
        //   <list-repetition> group <name> (LIST) {
        //     repeated group list {
        //                    ^~~~  repeatedGroupName
        //       <element-repetition> <element-type> element;
        //                                           ^~~~~~~  elementFieldName
        //     }
        //   }

        // Legacy mode, with non-nullable elements:
        //
        //   <list-repetition> group <name> (LIST) {
        //     repeated <element-type> array;
        //                             ^~~~~  repeatedFieldName
        //   }

[alexeykudinkin]

@wzx140 We should not be converting values to strings

[alexeykudinkin]

There's common annotation used for this: @VisibleForTesting

[alexeykudinkin]

Let's generalize this to log whatever record-type is used. Let's also downgrade to debug

[alexeykudinkin]

Why do we need this?

[wzx140]

The broadcast ensures that every executor has the same fingerprint->schema cache. We can't make every executor initialize fingerprint->schema locally.

For example, if we initialize fingerprint->schema in sparksqlwriter df, it is possible that the number of dataframe partitions is less than the number of executors.

        HoodieInternalRowUtils.addCompressedSchema(structType)
        df.queryExecution.toRdd.map(row => {
          val internalRow = row.copy()
          val (processedRow, writeSchema) = getSparkProcessedRecord(partitionCols, internalRow, dropPartitionColumns, structTypeBC.value)
          val recordKey = sparkKeyGenerator.getRecordKey(internalRow, structTypeBC.value)
          val partitionPath = sparkKeyGenerator.getPartitionPath(internalRow, structTypeBC.value)
          val key = new HoodieKey(recordKey.toString, partitionPath.toString)
          // See this We can't guarantee that this is executed in all executors
          HoodieInternalRowUtils.addCompressedSchema(structType)

          new HoodieSparkRecord(key, processedRow, writeSchema)
        }).toJavaRDD().asInstanceOf[JavaRDD[HoodieRecord[_]]]

As shown in the figure below, executor3 do not has the fingerPrint cache and can not deserialize the fingerPrint to schema.

[alexeykudinkin]

Sorry, i misread your original intent with this change. Unfortunately, this is not going to work: for Broadcast to work exactly the same object has to be passed in (w/in captured Closure) from Driver to Executor (Broadcasts are identified by unique bid). In the current setup you're just overwriting static variables w/in HoodieInternalRowUtils creating new Broadcasts which aren't propagated to Executors (the reason it works in tests is b/c we run phony Spark cluster where both Driver and Executor share memory, ie they access the same loaded HoodieInternalRowUtils and this problem is not apparent)

[alexeykudinkin]

We need a simpler solution here:

• We should remove the schema from HoodieSparkRecords. At this point we ultimately are familiar with all the limitations of the approach which Spark uses to reduce overhead of serializing the schema along w/ every record -- it requires us to provide all the schemas upfront to be distributed to all Executors (via SparkConf which is read-only after SparkContext is created).

• We will have to pass record's schema externally: we can approach it the same way it's currently done for ExpressionPayload in this PR

[wzx140]

I don't agree with that Broadcast does not work. I have tested this on yarn and it worked. The process of broadcast: the driver will put the blockcast variable into the blockmanager. And executors will be obtained locally according to the block id, and if it is not available locally, it will be pulled from the remote driver.

If the Broadcast works, we do not need to pass record's schema externally right? Passing record's schema externally requires user of HoodieRecord to pass the right schema.

[alexeykudinkin]

Leaving comments here for generateRecordMerger as well:

• Let's do config parsing here, passing in Seq[String] instead of String
• Let's rename generateRecordMerger to createRecordMerger

[alexeykudinkin]

Please downgrade to debug

[alexeykudinkin]

Why do we need this method?
Schema registration should be adhoc -- whenever someone is requesting to serialize HoodieSparkRecord we'd cache the schema, replacing w/ the fingerprint

[wzx140]

Removed.

[wzx140]

@hudi-bot run azure

[wzx140]

@alexeykudinkin I have implement without the Broadcast and schema inside HoodieSparkRecord and pass record's schema externally. Please review the update when you have time.

[wzx140]

@hudi-bot run azure

[wzx140]

CI fail with testReadFilterExistAfterBulkInsertPrepped. And it will fix after rebasing on master with commit HUDI-4923.

[alexeykudinkin]

@wzx140 would need to clarify this one

[alexeykudinkin]

I think we should invert this to be Option<Pair<HoodieRecord, Schema>>

[alexeykudinkin]

nit: hoodieRecord > newRecord

[alexeykudinkin]

@wzx140 this API is rather confusing: why do we need to pass all 3 records at the same time?

[alexeykudinkin]

Disregard. This is how it's right now, left note to self to revisit this.

[alexeykudinkin]

I see what you're working around now.

Let's leave a TODO and create a ticket to follow-up

• Remove record deflation: instead of deflation, we should just make sure we avoid keeping refs to the Record instant itself (so that it'd be collected)
• Remove these unnecessary newInstance invocations

[alexeykudinkin]

Right, but you're extracting this meta-fields from the current record, and so if it doesn't currently have it there's nothing to extract.
My point is we can't apply new schema to the record, until we rewrite the record into new schema -- we should use recordSchema to check whether it has the meta-fields

[alexeykudinkin]

@wzx140 let me elaborate on the context of my question:

I'm aware that Spark is using mutable buffers for reading the data, but we shouldn't need to copy it right away, we should only have to copy where absolutely necessary. Meaning that by default, if we'd just be iterating over the records from one file and writing it into another one we wouldn't need to make any copies, right? We can easily reuse the same buffer, and while we continue to read and write 1 record at a time (while iterating over provided Iterator).

val iter = parquet.read(file)
while (iter.hasNext) {
  writeToFile(iter.next)
}

We only really need to make a copy whenever we retain a reference to the Row, for ex, if we do reduceByKey (as below, or if we'd hold it in a Map, etc)

val iter = parquet.read(file)
iter.reduceByKey((r1, r2) => {
  // NOTE: In this case we're holding refs to 2 rows simultaneously, hence we'd need to make a copy before that
  // since otherwise both r1 and r2 will be pointing to the same buffer
})

So, we basically need:

• Avoid blanket copying here
• Find the place where we hold the refs to the row (as described above, this is the reason the test is failing) and allow copying in the least possible scope

Does it make sense?

[alexeykudinkin]

Sorry, my 2d statement was confusing, let me elaborate on my comment:

• This API converts from one HoodieRecord to another HoodieRecord
• There's no reason why we'd be returning an Option (from external perspective, it looks really strange that we actually return an Option, though we simply do a lateral conversion)
• Ideally we should be returning an Option when this object is dereferenced (getData is called), but this would affect the API

Let's keep it as it is for now, we can change it later when we'll be getting rid of HoodieAvroRecord

[alexeykudinkin]

Let's move this util to HoodieAvroRecordMerger (since it's only useful with it)

[alexeykudinkin]

@wzx140 we should actually produce HoodieAvroRecord here instead:

• Merged records will be subsequently reshuffled a few times, and shuffling IndexedRecord will kill performance
• Instead we should use HoodieAvroRecord holding some dummy RecordPayload just holds Avro bytes for us (and throws if anyone will try to use it for merging)

Does it make sense?

[wzx140]

I don't think so. In HoodieAvroRecordMerger, we do combineAndGetUpdateValue before writing and shuffling is done before combineAndGetUpdateValue. We do preCombine with record which hold avro byte and then shuffle the result. So there is no performce loss. This is also the original logic.

[alexeykudinkin]

You're right, i double-checked and we do indeed not shuffle the HoodieAvroIndexedRecord

[alexeykudinkin]

@wzx140 please create a ticket under RFC-46 epic, and link it here like TODO(HUDI-xxx)

[alexeykudinkin]

It's better to actually do the other way around:

• Keep inherited method named deduplicateRecords
• Create new private method deduplicateRecordsInternal modifying props and invoke it in this class

[wzx140]

Well, deduplicateRecords also used in StreamWriteFunction in Flink. So we should modify props in deduplicateRecords.

[alexeykudinkin]

@wzx140 you extract meta-fields from the current records using targetStructType which is a new schema.

Let's consider the case when current record doesn't have meta-fields (it doesn't have it in the schema), this code would fail unpredictably in that case since we're using field ordinals from targetStructType (targetStructType != structType)

[alexeykudinkin]

@wzx140 that's exactly what i'm referring to -- let's move such copying to the least possible scope to make sure we're only making copies where we absolutely have (and nowhere else)

[alexeykudinkin]

You're right, i double-checked and we do indeed not shuffle the HoodieAvroIndexedRecord

[alexeykudinkin]

Why are we changing this one?

[wzx140]

ParquetReaderIterator#next will return the current row and get the next row. We should get rid of it if we do not want to copy row in ParquetReaderIterator.

  T retVal = this.next;
  // This is what I want to remove if we not copy the retVal.
  this.next = read();
  return retVal;

[alexeykudinkin]

This is wrong, we should not be doing read() in next, instead it should be like following:

def hasNext() {
  if (next == null) {
    next = read()
  }
  return next != null
}

def next() {
  if (next == null) throw 
  val ret = next
  next = null
  return next
}

[wzx140]

@hudi-bot run azure

[hudi-bot]

CI report:

• 0650eb4 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build