Addressing feedback

Author: nsivabalan

Diff for: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java

@@ -296,41 +296,6 @@ private static ArrayList<ArrayList<String>> partitionKeysByFileSlices(List<Strin
     return partitionedKeys;
   }
 
-  /*@Override
-  public Map<String, List<HoodieRecord<HoodieMetadataPayload>>> getAllRecordsByKeys(List<String> keys, String partitionName) {
-    if (keys.isEmpty()) {
-      return Collections.emptyMap();
-    }
-
-    Map<String, List<HoodieRecord<HoodieMetadataPayload>>> result;
-    // Load the file slices for the partition. Each file slice is a shard which saves a portion of the keys.
-    List<FileSlice> partitionFileSlices = partitionFileSliceMap.computeIfAbsent(partitionName,
-        k -> HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(metadataMetaClient, getMetadataFileSystemView(), partitionName));
-    final int numFileSlices = partitionFileSlices.size();
-    checkState(numFileSlices > 0, "Number of file slices for partition " + partitionName + " should be > 0");
-
-    // Lookup keys from each file slice
-    if (numFileSlices == 1) {
-      // Optimization for a single slice for smaller metadata table partitions
-      result = lookupAllKeysFromFileSlice(partitionName, keys, partitionFileSlices.get(0));
-    } else {
-      // Parallel lookup for large sized partitions with many file slices
-      // Partition the keys by the file slice which contains it
-      ArrayList<ArrayList<String>> partitionedKeys = partitionKeysByFileSlices(keys, numFileSlices);
-      result = new HashMap<>(keys.size());
-      getEngineContext().setJobStatus(this.getClass().getSimpleName(), "Reading keys from metadata table partition " + partitionName);
-      getEngineContext().map(partitionedKeys, keysList -> {
-        if (keysList.isEmpty()) {
-          return Collections.<String, HoodieRecord<HoodieMetadataPayload>>emptyMap();
-        }
-        int shardIndex = HoodieTableMetadataUtil.mapRecordKeyToFileGroupIndex(keysList.get(0), numFileSlices);
-        return lookupAllKeysFromFileSlice(partitionName, keysList, partitionFileSlices.get(shardIndex));
-      }, partitionedKeys.size()).forEach(map -> result.putAll((Map<String, List<HoodieRecord<HoodieMetadataPayload>>>) map));
-    }
-
-    return result;
-  }*/
-
   /**
    * Lookup list of keys from a single file slice.
    *
@@ -439,119 +404,6 @@ private Map<String, HoodieRecord<HoodieMetadataPayload>> fetchBaseFileRecordsByK
     return result;
   }
 
-  /*private Map<String, List<HoodieRecord<HoodieMetadataPayload>>> lookupAllKeysFromFileSlice(String partitionName, List<String> keys, FileSlice fileSlice) {
-    Pair<HoodieSeekingFileReader<?>, HoodieMetadataLogRecordReader> readers = getOrCreateReaders(partitionName, fileSlice);
-    try {
-      List<Long> timings = new ArrayList<>();
-      HoodieSeekingFileReader<?> baseFileReader = readers.getKey();
-      HoodieMetadataLogRecordReader logRecordScanner = readers.getRight();
-      if (baseFileReader == null && logRecordScanner == null) {
-        return Collections.emptyMap();
-      }
-
-      // Sort it here once so that we don't need to sort individually for base file and for each individual log files.
-      List<String> sortedKeys = new ArrayList<>(keys);
-      Collections.sort(sortedKeys);
-      Map<String, List<HoodieRecord<HoodieMetadataPayload>>> logRecords = readAllLogRecords(logRecordScanner, sortedKeys, timings);
-      return readFromBaseAndMergeWithAllLogRecords(baseFileReader, sortedKeys, true, logRecords, timings, partitionName);
-    } catch (IOException ioe) {
-      throw new HoodieIOException("Error merging records from metadata table for  " + keys.size() + " key : ", ioe);
-    } finally {
-      if (!reuse) {
-        closeReader(readers);
-      }
-    }
-  }*/
-
-  /*private Map<String, List<HoodieRecord<HoodieMetadataPayload>>> readAllLogRecords(HoodieMetadataLogRecordReader logRecordReader,
-                                                                                   List<String> sortedKeys,
-                                                                                   List<Long> timings) {
-    HoodieTimer timer = HoodieTimer.start();
-
-    if (logRecordReader == null) {
-      timings.add(timer.endTimer());
-      return Collections.emptyMap();
-    }
-
-    try {
-      return logRecordReader.getAllRecordsByKeys(sortedKeys);
-    } finally {
-      timings.add(timer.endTimer());
-    }
-  }*/
-
-  /*private Map<String, List<HoodieRecord<HoodieMetadataPayload>>> readFromBaseAndMergeWithAllLogRecords(HoodieSeekingFileReader<?> reader,
-                                                                                                       List<String> sortedKeys,
-                                                                                                       boolean fullKeys,
-                                                                                                       Map<String, List<HoodieRecord<HoodieMetadataPayload>>> logRecords,
-                                                                                                       List<Long> timings,
-                                                                                                       String partitionName) throws IOException {
-    HoodieTimer timer = HoodieTimer.start();
-
-    if (reader == null) {
-      // No base file at all
-      timings.add(timer.endTimer());
-      return logRecords;
-    }
-
-    HoodieTimer readTimer = HoodieTimer.start();
-
-    Map<String, List<HoodieRecord<HoodieMetadataPayload>>> records =
-        fetchBaseFileAllRecordsByKeys(reader, sortedKeys, fullKeys, partitionName);
-
-    metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.BASEFILE_READ_STR, readTimer.endTimer()));
-
-    // Iterate over all provided log-records, merging them into existing records
-
-    logRecords.entrySet().forEach(kv -> {
-      records.merge(
-          kv.getKey(),
-          kv.getValue(),
-          (oldRecordList, newRecordList) -> {
-            List<HoodieRecord<HoodieMetadataPayload>> mergedRecordList = new ArrayList<>();
-            HoodieMetadataPayload mergedPayload = null;
-            HoodieKey key = null;
-            if (!oldRecordList.isEmpty() && !newRecordList.isEmpty()) {
-              mergedPayload = newRecordList.get(0).getData().preCombine(oldRecordList.get(0).getData());
-              key = newRecordList.get(0).getKey();
-            } else if (!oldRecordList.isEmpty()) {
-              mergedPayload = oldRecordList.get(0).getData();
-              key = oldRecordList.get(0).getKey();
-            } else if (!newRecordList.isEmpty()) {
-              mergedPayload = newRecordList.get(0).getData();
-              key = newRecordList.get(0).getKey();
-            }
-
-            if (mergedPayload != null && !mergedPayload.isDeleted()) {
-              mergedRecordList.add(new HoodieAvroRecord<>(key, mergedPayload));
-            }
-            return mergedRecordList;
-          }
-      );
-    });
-
-    timings.add(timer.endTimer());
-    return records;
-  }*/
-
-  /*private Map<String, List<HoodieRecord<HoodieMetadataPayload>>> fetchBaseFileAllRecordsByKeys(HoodieSeekingFileReader reader,
-                                                                                               List<String> sortedKeys,
-                                                                                               boolean fullKeys,
-                                                                                               String partitionName) throws IOException {
-    ClosableIterator<HoodieRecord<?>> records = fullKeys
-        ? reader.getRecordsByKeysIterator(sortedKeys)
-        : reader.getRecordsByKeyPrefixIterator(sortedKeys);
-
-    return toStream(records)
-        .map(record -> {
-          GenericRecord data = (GenericRecord) record.getData();
-          return Pair.of(
-              (String) (data).get(HoodieMetadataPayload.KEY_FIELD_NAME),
-              composeRecord(data, partitionName));
-        })
-        .collect(Collectors.groupingBy(Pair::getKey, Collectors.mapping(Pair::getValue, Collectors.toList())));
-  }*/
-
   private HoodieRecord<HoodieMetadataPayload> composeRecord(GenericRecord avroRecord, String partitionName) {
     if (metadataTableConfig.populateMetaFields()) {
       return SpillableMapUtils.convertToHoodieRecordPayload(avroRecord,

Diff for: hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/HFileUtils.java

@@ -200,11 +200,11 @@ public byte[] serializeRecordsToLogBlock(HoodieStorage storage,
 
       final byte[] recordBytes = serializeRecord(record, writerSchema, keyFieldName);
       if (sortedRecordsMap.containsKey(recordKey)) {
-        LOG.error("Found duplicate record with recordKey: " + recordKey);
+        LOG.error("Found duplicate record with recordKey: {} ", recordKey);
         printRecord("Previous record", sortedRecordsMap.get(recordKey), writerSchema);
         printRecord("Current record", recordBytes, writerSchema);
-        throw new HoodieException(String.format("Writing multiple records with same key %s not supported for %s",
-            recordKey, this.getClass().getName()));
+        throw new HoodieException(String.format("Writing multiple records with same key %s not supported for Hfile format with Metadata table",
+            recordKey));
       }
       sortedRecordsMap.put(recordKey, recordBytes);
     }
@@ -233,12 +233,16 @@ public byte[] serializeRecordsToLogBlock(HoodieStorage storage,
   }
 
   /**
-   * Print the record in json format
+   * Print the meta fields of the record of interest
    */
   private void printRecord(String msg, byte[] bs, Schema schema) throws IOException {
     GenericRecord record = HoodieAvroUtils.bytesToAvro(bs, schema);
-    byte[] json = HoodieAvroUtils.avroToJson(record, true);
-    LOG.error(String.format("%s: %s", msg, new String(json)));
+    if (schema.getField(HoodieRecord.RECORD_KEY_METADATA_FIELD) != null) {
+      LOG.error(String.format("%s: Hudi meta field values -> Record key: %s, Partition Path: %s, FileName: %s, CommitTime: %s, CommitSeqNo: %s", msg,
+          record.get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString()), record.get(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString(),
+          record.get(HoodieRecord.FILENAME_METADATA_FIELD).toString(), record.get(HoodieRecord.COMMIT_TIME_METADATA_FIELD).toString(),
+          record.get(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD).toString());
+    }
   }
 
   @Override

Diff for: hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieAvroHFileWriter.java

@@ -72,15 +72,14 @@ public class HoodieAvroHFileWriter
   private final String instantTime;
   private final TaskContextSupplier taskContextSupplier;
   private final boolean populateMetaFields;
-  private final Schema schema;
   private final Option<Schema.Field> keyFieldSchema;
   private HFile.Writer writer;
   private String minRecordKey;
   private String maxRecordKey;
   private String prevRecordKey;
 
   // This is private in CacheConfig so have been copied here.
-  private static String DROP_BEHIND_CACHE_COMPACTION_KEY = "hbase.hfile.drop.behind.compaction";
+  private static final String DROP_BEHIND_CACHE_COMPACTION_KEY = "hbase.hfile.drop.behind.compaction";
 
   public HoodieAvroHFileWriter(String instantTime, StoragePath file, HoodieHFileConfig hfileConfig, Schema schema,
                                TaskContextSupplier taskContextSupplier, boolean populateMetaFields) throws IOException {
@@ -91,7 +90,6 @@ public HoodieAvroHFileWriter(String instantTime, StoragePath file, HoodieHFileCo
     this.isWrapperFileSystem = fs instanceof HoodieWrapperFileSystem;
     this.wrapperFs = this.isWrapperFileSystem ? Option.of((HoodieWrapperFileSystem) fs) : Option.empty();
     this.hfileConfig = hfileConfig;
-    this.schema = schema;
     this.keyFieldSchema = Option.ofNullable(schema.getField(hfileConfig.getKeyFieldName()));
 
     // TODO - compute this compression ratio dynamically by looking at the bytes written to the

Diff for: hudi-hadoop-common/src/main/java/org/apache/hudi/io/hadoop/HoodieHBaseAvroHFileReader.java

@@ -371,7 +371,8 @@ public IndexedRecord next() {
   }
 
   private static Option<IndexedRecord> fetchRecordByKeyInternal(HFileScanner scanner, String key, Schema writerSchema, Schema readerSchema) throws IOException {
-    KeyValue kv = new KeyValue(getUTF8Bytes(key), null, null, null);
+    byte[] keyBytes = getUTF8Bytes(key);
+    KeyValue kv = new KeyValue(keyBytes, null, null, null);
     // NOTE: HFile persists both keys/values as bytes, therefore lexicographical sorted is
     //       essentially employed
     //
@@ -394,7 +395,7 @@ private static Option<IndexedRecord> fetchRecordByKeyInternal(HFileScanner scann
     // key is found and the cursor is left where the key is found
     Cell c = scanner.getCell();
     byte[] valueBytes = copyValueFromCell(c);
-    GenericRecord record = deserialize(getUTF8Bytes(key), valueBytes, writerSchema, readerSchema);
+    GenericRecord record = deserialize(keyBytes, valueBytes, writerSchema, readerSchema);
 
     return Option.of(record);
   }

Diff for: hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java

@@ -1980,9 +1980,6 @@ public void testColStatsPrefixLookup() throws IOException {
 
       // there are 3 partitions in total and 2 commits. total entries should be 6.
       assertEquals(result.size(), 6);
-      result.forEach(entry -> {
-        //LOG.warn("Prefix search entries just for record key col : " + entry.getRecordKey().toString() + " :: " + entry.getData().getColumnStatMetadata().get().toString());
-      });
 
       // prefix search for col(_hoodie_record_key) and first partition. only 2 files should be matched
       PartitionIndexID partitionIndexID = new PartitionIndexID(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH);
@@ -1991,7 +1988,6 @@ public void testColStatsPrefixLookup() throws IOException {
       // 1 partition and 2 commits. total entries should be 2.
       assertEquals(result.size(), 2);
       result.forEach(entry -> {
-        // LOG.warn("Prefix search entries for record key col and first partition : " + entry.getRecordKey().toString() + " :: " + entry.getData().getColumnStatMetadata().get().toString());
         HoodieMetadataColumnStats metadataColumnStats = entry.getData().getColumnStatMetadata().get();
         String fileName = metadataColumnStats.getFileName();
         if (fileName.contains(firstCommit)) {
@@ -2011,7 +2007,6 @@ public void testColStatsPrefixLookup() throws IOException {
       // 1 partition and 2 commits. total entries should be 2.
       assertEquals(result.size(), 2);
       result.forEach(entry -> {
-        // LOG.warn("Prefix search entries for record key col and first partition : " + entry.getRecordKey().toString() + " :: " + entry.getData().getColumnStatMetadata().get().toString());
         HoodieMetadataColumnStats metadataColumnStats = entry.getData().getColumnStatMetadata().get();
         // for commit time column, min max should be the same since we disable small files, every commit will create a new file
         assertEquals(metadataColumnStats.getMinValue(), metadataColumnStats.getMaxValue());